From 1b5c7b3b1ed7c1c12c794ba408b064104cdaf2ec Mon Sep 17 00:00:00 2001 From: Krish Gandhi Date: Mon, 27 Jul 2026 14:36:05 -0700 Subject: [PATCH 1/4] Adding cache hit analysis, cache disabling, recommendations to analyze Signed-off-by: Krish Gandhi --- .../sql/calcite/CalcitePlanContext.java | 4 + .../sql/executor/AnalyzeResponse.java | 20 +- .../sql/executor/ExecutionEngine.java | 8 + .../opensearch/sql/executor/QueryService.java | 567 +++++++++++++++--- .../org/opensearch/sql/storage/Table.java | 9 + .../opensearch/client/OpenSearchClient.java | 8 + .../client/OpenSearchNodeClient.java | 11 + .../client/OpenSearchRestClient.java | 13 + .../executor/OpenSearchExecutionEngine.java | 24 + .../request/OpenSearchQueryRequest.java | 13 + .../opensearch/storage/OpenSearchIndex.java | 5 + 11 files changed, 607 insertions(+), 75 deletions(-) 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..5a17a2102f2 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 skipEncoding = ThreadLocal.withInitial(() -> false); + /** When true, disables the OpenSearch shard request cache for the current query. */ + public static final ThreadLocal disableRequestCache = + ThreadLocal.withInitial(() -> false); + /** When true, the execution engine strips all-null columns from the result (used by timewrap). */ public static final ThreadLocal stripNullColumns = ThreadLocal.withInitial(() -> false); 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..f714b342fa7 100644 --- a/core/src/main/java/org/opensearch/sql/executor/AnalyzeResponse.java +++ b/core/src/main/java/org/opensearch/sql/executor/AnalyzeResponse.java @@ -21,11 +21,12 @@ public class AnalyzeResponse { private final List physicalPlan; private final QueryProfile profile; private final List operator_tree; - private final List recommendations; + private final List recommendations; private final List schema; private final Object[][] datarows; private final long total; private final long size; + private final boolean possibleCacheHit; @Data @Builder @@ -46,6 +47,7 @@ public static class QuerySegment { public static class OperatorNode { private final String source; private final List node_type; + private final List node_cost; private final List description; private final String estimated_cost; private final Long estimated_rows; @@ -53,4 +55,20 @@ public static class OperatorNode { private final Long actual_rows; private final Boolean is_pushed_down; } + + public enum RecommendationSeverityLevel { + INFO, + WARNING, + CRITICAL + } + + @Data + @Builder + public static class Recommendation { + private final RecommendationSeverityLevel serverity; + 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..b098af79763 100644 --- a/core/src/main/java/org/opensearch/sql/executor/QueryService.java +++ b/core/src/main/java/org/opensearch/sql/executor/QueryService.java @@ -24,27 +24,23 @@ import org.apache.calcite.plan.RelTraitDef; import org.apache.calcite.rel.RelCollation; import org.apache.calcite.rel.RelCollations; -import org.apache.calcite.rel.RelHomogeneousShuttle; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.RelRoot; import org.apache.calcite.rel.core.Sort; import org.apache.calcite.rel.logical.LogicalSort; -import org.apache.calcite.rex.RexCall; -import org.apache.calcite.rex.RexNode; -import org.apache.calcite.rex.RexShuttle; import org.apache.calcite.runtime.Hook; import org.apache.calcite.schema.SchemaPlus; import org.apache.calcite.sql.SqlExplainLevel; -import org.apache.calcite.sql.SqlOperator; -import org.apache.calcite.sql.fun.SqlStdOperatorTable; import org.apache.calcite.sql.parser.SqlParser; import org.apache.calcite.tools.FrameworkConfig; import org.apache.calcite.tools.Frameworks; 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; @@ -224,9 +220,7 @@ public void executeWithCalcite( RelNode calcitePlan = StageErrorHandler.executeStage( QueryProcessingStage.PLAN_CONVERSION, - () -> - withCheckedArithmetic( - convertToCalcitePlan(relNode, context), context), + () -> convertToCalcitePlan(relNode, context), "while converting the query to an executable plan"); executeCalcitePlan(calcitePlan, context, listener, analyzeMetric, analyzeStart); @@ -301,8 +295,7 @@ public void explainWithCalcite( context.run( () -> { RelNode relNode = analyze(plan, context); - RelNode calcitePlan = - withCheckedArithmetic(convertToCalcitePlan(relNode, context), context); + RelNode calcitePlan = convertToCalcitePlan(relNode, context); if (format != null) { executionEngine.explain(calcitePlan, mode, format, context, listener); } else { @@ -328,7 +321,7 @@ public void analyzeWithCalcite( String query, List querySegments, UnresolvedPlan plan, - QueryType queryType, + QueryType queryType, // boolean disableCache, ResponseListener listener) { if (!shouldUseCalcite(queryType)) { listener.onFailure( @@ -337,10 +330,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 queryResponseRef = new AtomicReference<>(); AtomicReference profileRef = new AtomicReference<>(); AtomicReference errorRef = new AtomicReference<>(); @@ -379,8 +381,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,6 +393,13 @@ 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(); @@ -415,8 +427,9 @@ public void onFailure(Exception e) { } listener.onResponse( AnalyzeResponse.builder() - // .query(query) + .query(query) .profile(profile) + .possibleCacheHit(possibleCacheHit) .schema(schema) .datarows(datarows) .total(datarows.length) @@ -502,6 +515,43 @@ public void onFailure(Exception e) { .toArray(Object[]::new); } + // Extract scan metadata for recommendations #2 and #3. + org.apache.calcite.rel.metadata.RelMetadataQuery mq = + calcitePlan.getCluster().getMetadataQuery(); + long totalIndexDocs = getIndexDocCount(plan, context); + if (totalIndexDocs <= 0) { + totalIndexDocs = getScanBaseRowCount(calcitePlan, mq); + } + Set dateFieldNames = getDateFieldNames(plan, context); + boolean isTimeSeriesIndex = + !dateFieldNames.isEmpty() + || hasDateField(calcitePlan) + || hasDateFieldByName(logicalPlanNodes); + boolean hasDateRangeFilter = + logicalPlanNodes.stream() + .anyMatch( + n -> + n.contains("LogicalFilter") + && dateFieldNames.stream() + .anyMatch( + f -> + n.contains(f) + || n.contains("TIMESTAMP") + || n.contains("date("))) + || physicalPlanNodes.stream() + .anyMatch( + n -> + n.contains("FILTER") + && dateFieldNames.stream().anyMatch(n::contains)); + + List recommendations = + buildRecommendations( + operatorTree, + profile, + totalIndexDocs, + isTimeSeriesIndex, + hasDateRangeFilter); + AnalyzeResponse response = AnalyzeResponse.builder() .query(query) @@ -509,8 +559,9 @@ public void onFailure(Exception e) { .logicalPlan(logicalPlanNodes) .physicalPlan(physicalPlanNodes) .operator_tree(operatorTree) - .recommendations(List.of()) + .recommendations(recommendations) .profile(profile) + .possibleCacheHit(possibleCacheHit) .schema(schema) .datarows(datarows) .total(datarows.length) @@ -606,6 +657,10 @@ private List buildOperatorTree( Map idToRowCount = new HashMap<>(); collectRowCounts(logicalPlan, mq, idToRowCount); + // Collect non-cumulative costs per logical node for cost attribution. + Map idToCost = new HashMap<>(); + collectNonCumulativeCosts(logicalPlan, mq, idToCost); + // 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 physicalTimings = new ArrayList<>(); @@ -630,6 +685,17 @@ private List buildOperatorTree( } } + // Collect per-segment plan IDs for cost attribution across all segments. + List> allSegmentPlanIds = new ArrayList<>(); + for (int i = 0; i < querySegments.size(); i++) { + Set ids = i < exclusiveIds.size() ? exclusiveIds.get(i) : Set.of(); + allSegmentPlanIds.add( + ids.stream() + .filter(idToDescription::containsKey) + .collect(java.util.stream.Collectors.toSet())); + } + List allSegmentCosts = computeSegmentCosts(allSegmentPlanIds, idToCost); + List operators = new ArrayList<>(); int physicalIdx = 0; @@ -652,6 +718,7 @@ private List buildOperatorTree( .orElse(""); List nodeTypes = mergedSegments.stream().map(AnalyzeResponse.QuerySegment::getNodeType).toList(); + List nodeCosts = allSegmentCosts.subList(0, pushedSegments); // Collect all plan node ids in the pushed group for estimated_rows Set allPushedPlanIds = new HashSet<>(); for (int i = 0; i < pushedSegments; i++) { @@ -665,6 +732,7 @@ private List buildOperatorTree( AnalyzeResponse.OperatorNode.builder() .source(combinedSource) .node_type(nodeTypes) + .node_cost(nodeCosts) .description(descriptions.isEmpty() ? null : descriptions) .is_pushed_down(true) .estimated_rows(getEstimatedRows(allPushedPlanIds, idToRowCount)) @@ -687,6 +755,7 @@ private List buildOperatorTree( AnalyzeResponse.OperatorNode.builder() .source(seg.getSource()) .node_type(List.of(seg.getNodeType())) + .node_cost(List.of(allSegmentCosts.get(0))) .description(descriptions.isEmpty() ? null : descriptions) .estimated_rows(getEstimatedRows(planIds, idToRowCount)) .actual_time_ms(timing != null ? String.format("%.2f ms", timing[0]) : null) @@ -703,9 +772,11 @@ private List buildOperatorTree( List group = new ArrayList<>(); List descriptions = new ArrayList<>(); Set groupPlanIds = new HashSet<>(); + List groupCosts = new ArrayList<>(); long logicalNodesInGroup = 0; while (idx < querySegments.size() && logicalNodesInGroup < 1) { group.add(querySegments.get(idx)); + groupCosts.add(allSegmentCosts.get(idx)); Set ids = idx < exclusiveIds.size() ? exclusiveIds.get(idx) : Set.of(); ids.stream() .sorted() @@ -730,6 +801,7 @@ private List buildOperatorTree( AnalyzeResponse.OperatorNode.builder() .source(combinedSource) .node_type(nodeTypes) + .node_cost(groupCosts) .description(descriptions.isEmpty() ? null : descriptions) .estimated_rows(getEstimatedRows(groupPlanIds, idToRowCount)) .actual_time_ms(timing != null ? String.format("%.2f ms", timing[0]) : null) @@ -765,6 +837,178 @@ private static int getLinearDepth(RelNode node) { return depth; } + private static RelNode getLeafScanNode(RelNode node) { + RelNode current = node; + while (current != null) { + List inputs = current.getInputs(); + if (inputs.isEmpty()) { + return current; + } + current = inputs.get(0); + } + return node; + } + + private static long getIndexDocCount(UnresolvedPlan plan, CalcitePlanContext context) { + try { + String[] indexNames = extractIndexNames(plan); + if (indexNames.length == 0) { + return -1; + } + org.apache.calcite.schema.SchemaPlus schema = context.config.getDefaultSchema(); + org.apache.calcite.schema.Table calciteTable = schema.getTable(indexNames[0]); + if (calciteTable instanceof org.opensearch.sql.storage.Table storageTable) { + return storageTable.getDocCount(); + } + } catch (Exception ignored) { + } + return -1; + } + + private static long getScanBaseRowCount( + RelNode plan, org.apache.calcite.rel.metadata.RelMetadataQuery mq) { + RelNode leaf = getLeafScanNode(plan); + try { + Double rowCount = mq.getRowCount(leaf); + if (rowCount != null) { + return Math.round(rowCount); + } + } catch (Exception ignored) { + } + return -1; + } + + private static boolean hasDateField(RelNode plan) { + RelNode leaf = getLeafScanNode(plan); + try { + org.apache.calcite.rel.type.RelDataType rowType; + if (leaf instanceof org.apache.calcite.rel.core.TableScan tableScan) { + rowType = tableScan.getTable().getRowType(); + } else { + rowType = leaf.getRowType(); + } + for (org.apache.calcite.rel.type.RelDataTypeField field : rowType.getFieldList()) { + org.apache.calcite.rel.type.RelDataType fieldType = field.getType(); + org.apache.calcite.sql.type.SqlTypeName typeName = fieldType.getSqlTypeName(); + if (typeName == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP + || typeName == org.apache.calcite.sql.type.SqlTypeName.DATE + || typeName == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE) { + return true; + } + if (fieldType + instanceof org.opensearch.sql.calcite.type.AbstractExprRelDataType exprType) { + org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT udt = exprType.getUdt(); + if (udt == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_TIMESTAMP + || udt == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_DATE) { + return true; + } + } + } + } catch (Exception ignored) { + } + return false; + } + + /** + * Checks whether the index backing the query has a date/timestamp field by looking up the full + * table schema from the Calcite schema registry. Unlike {@link #hasDateField(RelNode)}, this + * approach is not affected by project pushdown which narrows the scan's row type to only the + * fields referenced in the query. + */ + private static boolean hasDateFieldFromSchema(UnresolvedPlan plan, CalcitePlanContext context) { + try { + String[] indexNames = extractIndexNames(plan); + if (indexNames.length == 0) { + return false; + } + org.apache.calcite.schema.SchemaPlus schema = context.config.getDefaultSchema(); + for (String indexName : indexNames) { + org.apache.calcite.schema.Table table = schema.getTable(indexName); + if (table == null) { + continue; + } + org.apache.calcite.rel.type.RelDataType fullRowType = + table.getRowType(org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.TYPE_FACTORY); + for (org.apache.calcite.rel.type.RelDataTypeField field : fullRowType.getFieldList()) { + org.apache.calcite.rel.type.RelDataType fieldType = field.getType(); + org.apache.calcite.sql.type.SqlTypeName typeName = fieldType.getSqlTypeName(); + if (typeName == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP + || typeName == org.apache.calcite.sql.type.SqlTypeName.DATE + || typeName + == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE) { + return true; + } + if (fieldType + instanceof org.opensearch.sql.calcite.type.AbstractExprRelDataType exprType) { + org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT udt = exprType.getUdt(); + if (udt == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_TIMESTAMP + || udt + == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_DATE) { + return true; + } + } + } + } + } catch (Exception ignored) { + } + return false; + } + + private static Set getDateFieldNames(UnresolvedPlan plan, CalcitePlanContext context) { + Set dateFields = new HashSet<>(); + try { + String[] indexNames = extractIndexNames(plan); + if (indexNames.length == 0) { + return dateFields; + } + org.apache.calcite.schema.SchemaPlus schema = context.config.getDefaultSchema(); + for (String indexName : indexNames) { + org.apache.calcite.schema.Table table = schema.getTable(indexName); + if (table == null) { + continue; + } + org.apache.calcite.rel.type.RelDataType fullRowType = + table.getRowType(org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.TYPE_FACTORY); + for (org.apache.calcite.rel.type.RelDataTypeField field : fullRowType.getFieldList()) { + org.apache.calcite.rel.type.RelDataType fieldType = field.getType(); + org.apache.calcite.sql.type.SqlTypeName typeName = fieldType.getSqlTypeName(); + if (typeName == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP + || typeName == org.apache.calcite.sql.type.SqlTypeName.DATE + || typeName + == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE) { + dateFields.add(field.getName()); + } else if (fieldType + instanceof org.opensearch.sql.calcite.type.AbstractExprRelDataType exprType) { + org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT udt = exprType.getUdt(); + if (udt == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_TIMESTAMP + || udt + == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_DATE) { + dateFields.add(field.getName()); + } + } + } + } + } catch (Exception ignored) { + } + return dateFields; + } + + private static boolean hasDateFieldByName(List logicalPlanNodes) { + for (String node : logicalPlanNodes) { + if (node.contains("timestamp") + || node.contains("@timestamp") + || node.contains("event_time") + || node.contains("created_at") + || node.contains("updated_at") + || node.contains("date_field") + || node.contains("EXPR_TIMESTAMP") + || node.contains("EXPR_DATE")) { + return true; + } + } + return false; + } + private void collectRowCounts( RelNode node, org.apache.calcite.rel.metadata.RelMetadataQuery mq, @@ -781,6 +1025,23 @@ private void collectRowCounts( } } + private void collectNonCumulativeCosts( + RelNode node, + org.apache.calcite.rel.metadata.RelMetadataQuery mq, + Map idToCost) { + try { + org.apache.calcite.plan.RelOptCost cost = mq.getNonCumulativeCost(node); + if (cost != null && !cost.isInfinite()) { + double weight = cost.getCpu() + cost.getIo() * 10.0 + cost.getRows(); + idToCost.put(node.getId(), weight); + } + } catch (Exception ignored) { + } + for (RelNode input : node.getInputs()) { + collectNonCumulativeCosts(input, mq, idToCost); + } + } + private Long getEstimatedRows(Set ids, Map idToRowCount) { return ids.stream() .filter(idToRowCount::containsKey) @@ -789,6 +1050,207 @@ private Long getEstimatedRows(Set ids, Map idToRowCoun .orElse(null); } + /** + * Compute per-segment cost fractions from Calcite's non-cumulative cost. Each segment's cost is + * the sum of its exclusive RelNode costs, normalized to a percentage of the total across all + * segments in the operator tree. + */ + private List computeSegmentCosts( + List> segmentPlanIds, Map idToCost) { + List rawCosts = new ArrayList<>(); + for (Set ids : segmentPlanIds) { + double segCost = ids.stream().filter(idToCost::containsKey).mapToDouble(idToCost::get).sum(); + rawCosts.add(segCost); + } + double total = rawCosts.stream().mapToDouble(Double::doubleValue).sum(); + if (total <= 0) { + return rawCosts.stream().map(c -> 0f).toList(); + } + return rawCosts.stream().map(c -> (float) (c / total * 100.0)).toList(); + } + + private List buildRecommendations( + List operatorTree, + QueryProfile profile, + long totalIndexDocs, + boolean isTimeSeriesIndex, + boolean hasDateRangeFilter) { + List recommendations = new ArrayList<>(); + if (operatorTree == null || operatorTree.isEmpty() || profile == null) { + return recommendations; + } + + QueryProfile.Phase executePhase = profile.getPhases().get("execute"); + if (executePhase == null || executePhase.getTimeMillis() <= 0) { + return recommendations; + } + double executeTime = executePhase.getTimeMillis(); + + double maxTime = 0; + AnalyzeResponse.OperatorNode bottleneck = null; + + for (AnalyzeResponse.OperatorNode node : operatorTree) { + if (node.getActual_time_ms() == null) { + continue; + } + double time = parseTimeMs(node.getActual_time_ms()); + if (time > maxTime) { + maxTime = time; + bottleneck = node; + } + } + + int totalNodes = operatorTree.size(); + int pushedDown = 0; + for (AnalyzeResponse.OperatorNode node : operatorTree) { + if (Boolean.TRUE.equals(node.getIs_pushed_down())) { + pushedDown++; + } + } + int inMemory = totalNodes - pushedDown; + if (totalNodes > 0) { + recommendations.add( + AnalyzeResponse.Recommendation.builder() + .serverity(AnalyzeResponse.RecommendationSeverityLevel.INFO) + .rule("Pushdown visibility") + .message( + pushedDown + + " of " + + totalNodes + + " stages pushed down; " + + inMemory + + " ran in-memory") + .build()); + } + + if (bottleneck != null && maxTime > 0) { + long pct = Math.round((maxTime / executeTime) * 100); + String stage = + (bottleneck.getNode_type() != null && !bottleneck.getNode_type().isEmpty()) + ? String.join(", ", bottleneck.getNode_type()) + : "unknown"; + recommendations.add( + AnalyzeResponse.Recommendation.builder() + .serverity(AnalyzeResponse.RecommendationSeverityLevel.INFO) + .rule("Bottleneck stage") + .message(pct + "% of time is in the *" + stage + "* stage") + .affected_node(bottleneck.getSource()) + .suggestion("Consider optimizing the " + stage + " operation") + .build()); + } + + // In-memory bottleneck: find the non-pushed-down node with the highest self-time + double maxInMemoryTime = 0; + AnalyzeResponse.OperatorNode inMemoryBottleneck = null; + for (AnalyzeResponse.OperatorNode node : operatorTree) { + if (Boolean.TRUE.equals(node.getIs_pushed_down())) { + continue; + } + if (node.getActual_time_ms() == null) { + continue; + } + double time = parseTimeMs(node.getActual_time_ms()); + if (time > maxInMemoryTime) { + maxInMemoryTime = time; + inMemoryBottleneck = node; + } + } + if (inMemoryBottleneck != null + && maxInMemoryTime > 0 + && inMemoryBottleneck.getActual_rows() != null) { + long pct = Math.round((maxInMemoryTime / executeTime) * 100); + String stage = + (inMemoryBottleneck.getNode_type() != null + && !inMemoryBottleneck.getNode_type().isEmpty()) + ? String.join(", ", inMemoryBottleneck.getNode_type()) + : "unknown"; + recommendations.add( + AnalyzeResponse.Recommendation.builder() + .serverity(AnalyzeResponse.RecommendationSeverityLevel.WARNING) + .rule("In-memory bottleneck") + .message( + "Your *" + + stage + + "* ran in-memory over " + + inMemoryBottleneck.getActual_rows() + + " rows (" + + pct + + "% of time)") + .affected_node(inMemoryBottleneck.getSource()) + .suggestion( + "Consider pushing this operation down or reducing input rows with filters") + .build()); + } + + // Low scan selectivity: scan rows / total index docs > 80% + log.info( + "Low scan selectivity check: totalIndexDocs={}, operatorTree.size={}", + totalIndexDocs, + operatorTree.size()); + if (totalIndexDocs > 0) { + AnalyzeResponse.OperatorNode scanNode = operatorTree.get(0); + log.info( + "Low scan selectivity: scanNode.actual_rows={}, scanNode.estimated_rows={}", + scanNode.getActual_rows(), + scanNode.getEstimated_rows()); + if (scanNode.getActual_rows() != null && scanNode.getActual_rows() > 0) { + long scannedRows = scanNode.getActual_rows(); + long pct = Math.round((double) scannedRows / totalIndexDocs * 100); + long resultRows = + operatorTree.get(operatorTree.size() - 1).getActual_rows() != null + ? operatorTree.get(operatorTree.size() - 1).getActual_rows() + : 0; + log.info( + "Low scan selectivity: scannedRows={}, pct={}, resultRows={}", + scannedRows, + pct, + resultRows); + if (pct > 80) { + recommendations.add( + AnalyzeResponse.Recommendation.builder() + .serverity(AnalyzeResponse.RecommendationSeverityLevel.WARNING) + .rule("Low scan selectivity") + .message( + "Scanned " + + scannedRows + + " docs (" + + pct + + "% of index) to return " + + resultRows + + " rows") + .affected_node(scanNode.getSource()) + .suggestion("Add filters to reduce the number of documents scanned") + .build()); + } + } + } + + // Missing time filter: time-series index with no date range predicate pushed down + if (isTimeSeriesIndex && !hasDateRangeFilter) { + AnalyzeResponse.OperatorNode scanNode = operatorTree.get(0); + recommendations.add( + AnalyzeResponse.Recommendation.builder() + .serverity(AnalyzeResponse.RecommendationSeverityLevel.CRITICAL) + .rule("Missing time filter") + .message("No time filter on a time-series index: add one") + .affected_node(scanNode.getSource()) + .suggestion( + "Add a time range filter (e.g. where @timestamp > now() - interval 1 hour)") + .build()); + } + + return recommendations; + } + + private static double parseTimeMs(String timeMsStr) { + String stripped = timeMsStr.replaceAll("[^0-9.]", ""); + try { + return Double.parseDouble(stripped); + } catch (NumberFormatException e) { + return 0; + } + } + public void executeWithLegacy( UnresolvedPlan plan, QueryType queryType, @@ -923,66 +1385,6 @@ private boolean isCalciteEnabled(Settings settings) { } } - /** - * Rewrite {@code +}/{@code -}/{@code *} to their overflow-checked variants ({@code CHECKED_PLUS} - * / {@code CHECKED_MINUS} / {@code CHECKED_MULTIPLY}) so integer and long arithmetic overflow - * throws {@link ArithmeticException} (via {@code Math.addExact} etc.) instead of silently - * wrapping. Applied before pushdown so both coordinator-executed and pushed-down (script) - * arithmetic are checked. Floating-point arithmetic is unchanged (IEEE 754). - * - *

This does the same rewrite as Calcite's {@code ConvertToChecked} but preserves each call's - * originally inferred type (via {@code makeCall(type, op, operands)}) and touches only the three - * arithmetic operators, so it does not re-derive the types of unrelated calls (e.g. {@code - * CEIL}/{@code DIVIDE}) the way {@code ConvertToChecked} does. - */ - private static RelNode withCheckedArithmetic(RelNode calcitePlan, CalcitePlanContext context) { - RexShuttle checkedShuttle = - new RexShuttle() { - @Override - public RexNode visitCall(RexCall call) { - RexNode visited = super.visitCall(call); - if (!(visited instanceof RexCall rexCall)) { - return visited; - } - SqlOperator checked = - switch (rexCall.getOperator().getKind()) { - case PLUS -> SqlStdOperatorTable.CHECKED_PLUS; - case MINUS -> SqlStdOperatorTable.CHECKED_MINUS; - case TIMES -> SqlStdOperatorTable.CHECKED_MULTIPLY; - default -> null; - }; - // Only integer/long arithmetic can overflow silently and has a checked - // implementation (Math.addExact etc.). Float/double/decimal have no checked variant - // (SqlFunctions.checkedMultiply(double,double) does not exist) and follow IEEE 754, so - // leave them untouched. - if (checked == null || !isCheckableIntegerArithmetic(rexCall)) { - return visited; - } - return context.rexBuilder.makeCall(rexCall.getType(), checked, rexCall.getOperands()); - } - }; - return calcitePlan.accept( - new RelHomogeneousShuttle() { - @Override - public RelNode visit(RelNode other) { - RelNode visited = super.visitChildren(other); - return visited.accept(checkedShuttle); - } - }); - } - - /** Returns whether the result and every operand are BIGINT. */ - private static boolean isCheckableIntegerArithmetic(RexCall call) { - if (!isCheckableLongType(call.getType())) { - return false; - } - return call.getOperands().stream().allMatch(op -> isCheckableLongType(op.getType())); - } - - private static boolean isCheckableLongType(org.apache.calcite.rel.type.RelDataType type) { - return type.getSqlTypeName() == org.apache.calcite.sql.type.SqlTypeName.BIGINT; - } - /** * Walk the cause chain to find an {@link ArithmeticException} raised by checked arithmetic. Row- * level overflow surfaces wrapped (SQLException -> RuntimeException -> ErrorReport), so a @@ -1021,6 +1423,23 @@ private FrameworkConfig buildFrameworkConfig() { return configBuilder.build(); } + private static String[] extractIndexNames(UnresolvedPlan plan) { + Set names = new HashSet<>(); + collectRelationNames(plan, names); + return names.toArray(String[]::new); + } + + private static void collectRelationNames(Node node, Set 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/storage/Table.java b/core/src/main/java/org/opensearch/sql/storage/Table.java index 33dbd7d66d3..13f3e86ff4b 100644 --- a/core/src/main/java/org/opensearch/sql/storage/Table.java +++ b/core/src/main/java/org/opensearch/sql/storage/Table.java @@ -35,6 +35,15 @@ default void create(Map schema) { throw new UnsupportedOperationException("Unsupported Operation"); } + /** + * Get the total document count for this table. + * + * @return total document count, or -1 if unavailable + */ + default long getDocCount() { + return -1; + } + /** Get the {@link ExprType} for each field in the table. */ Map getFieldTypes(); diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchClient.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchClient.java index 68350c5a0fd..a9d5b25c14f 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchClient.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchClient.java @@ -55,6 +55,14 @@ public interface OpenSearchClient { */ Map getIndexMaxResultWindows(String... indexExpression); + /** + * Get the total document count for the given index expression. + * + * @param indexExpression index expression + * @return total document count + */ + long getIndexDocCount(String indexExpression); + /** * Perform search query in the search request. * 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..0f76e981fe8 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 @@ -132,6 +132,17 @@ public Map getIndexMappings(String... indexExpression) { * @param indexExpression index expression * @return map from index name to its max result window */ + @Override + public long getIndexDocCount(String indexExpression) { + try { + org.opensearch.action.admin.indices.stats.IndicesStatsResponse response = + client.admin().indices().prepareStats(indexExpression).clear().setDocs(true).get(); + return response.getTotal().getDocs().getCount(); + } catch (Exception e) { + return -1; + } + } + @Override public Map getIndexMaxResultWindows(String... indexExpression) { try { diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchRestClient.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchRestClient.java index f369c0003b8..e841ad1a739 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchRestClient.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchRestClient.java @@ -88,6 +88,19 @@ public Map getIndexMappings(String... indexExpression) { } } + @Override + public long getIndexDocCount(String indexExpression) { + try { + org.opensearch.client.core.CountRequest countRequest = + new org.opensearch.client.core.CountRequest(indexExpression); + org.opensearch.client.core.CountResponse response = + client.count(countRequest, RequestOptions.DEFAULT); + return response.getCount(); + } catch (Exception e) { + return -1; + } + } + @Override public Map getIndexMaxResultWindows(String... indexExpression) { GetSettingsRequest request = 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 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 afterKey; + @EqualsAndHashCode.Exclude @ToString.Exclude private final boolean disableRequestCache; + @TestOnly static OpenSearchQueryRequest of( String indexName, int size, OpenSearchExprValueFactory factory, List 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 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 searchRequest = new SearchRequest().indices(indexName.getIndexNames()).source(this.sourceBuilder); + if (disableRequestCache) { + searchRequest.requestCache(false); + } this.searchResponse = searchAction.apply(searchRequest); openSearchResponse = diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java index 3350c00fb0c..1ab960da4e0 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java @@ -191,6 +191,11 @@ public Map getFieldOpenSearchTypes() { return cachedFieldOpenSearchTypes; } + @Override + public long getDocCount() { + return client.getIndexDocCount(indexName.toString()); + } + /** Get the max result window setting of the table. */ public Integer getMaxResultWindow() { if (cachedMaxResultWindow == null) { From ac2182c697d2675ba6468501aef5605232af6447 Mon Sep 17 00:00:00 2001 From: Krish Gandhi Date: Thu, 13 Aug 2026 14:37:06 -0700 Subject: [PATCH 2/4] Removing operator_tree and query op tracking, adding recommendations Signed-off-by: Krish Gandhi --- .../sql/calcite/CalcitePlanContext.java | 16 - .../sql/calcite/CalciteRelNodeVisitor.java | 34 - .../sql/executor/AnalyzeResponse.java | 29 +- .../opensearch/sql/executor/QueryService.java | 765 +----------------- .../analyze/AnalyzeRecommendationBuilder.java | 285 +++++++ .../sql/executor/execution/AnalyzePlan.java | 7 +- .../executor/execution/QueryPlanFactory.java | 5 +- .../AnalyzeRecommendationBuilderTest.java | 160 ++++ docs/user/ppl/interfaces/endpoint.md | 45 +- .../sql/calcite/remote/CalciteAnalyzeIT.java | 150 +--- .../rest-api-spec/test/api/ppl.analyze.yml | 26 +- .../rest-api-spec/test/api/ppl.profile.yml | 5 - .../transport/TransportPPLQueryAction.java | 2 +- .../org/opensearch/sql/ppl/PPLService.java | 55 +- .../ppl/calcite/CalcitePPLTrackingTest.java | 266 ------ 15 files changed, 461 insertions(+), 1389 deletions(-) create mode 100644 core/src/main/java/org/opensearch/sql/executor/analyze/AnalyzeRecommendationBuilder.java create mode 100644 core/src/test/java/org/opensearch/sql/executor/analyze/AnalyzeRecommendationBuilderTest.java delete mode 100644 ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLTrackingTest.java 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 5a17a2102f2..9ee2556668b 100644 --- a/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java +++ b/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java @@ -128,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 nodeIdMappings = new ArrayList<>(); - - /** Records a mapping from an AST command to the RelNode ids it produced. */ - public void recordMapping(String astNodeType, List 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 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 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 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 f714b342fa7..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,14 +13,10 @@ @Data @Builder public class AnalyzeResponse { - - private final String query; - private final List querySegments; - // private final String ast; + // private final String query; private final List logicalPlan; private final List physicalPlan; private final QueryProfile profile; - private final List operator_tree; private final List recommendations; private final List schema; private final Object[][] datarows; @@ -35,27 +31,6 @@ public static class SchemaColumn { private final String type; } - @Data - @Builder - public static class QuerySegment { - private final String nodeType; - private final String source; - } - - @Data - @Builder - public static class OperatorNode { - private final String source; - private final List node_type; - private final List node_cost; - private final List 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 enum RecommendationSeverityLevel { INFO, WARNING, @@ -65,7 +40,7 @@ public enum RecommendationSeverityLevel { @Data @Builder public static class Recommendation { - private final RecommendationSeverityLevel serverity; + private final RecommendationSeverityLevel severity; private final String rule; private final String message; private final String affected_node; 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 b098af79763..0b7a03629d7 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; @@ -60,6 +57,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; @@ -319,7 +317,6 @@ public void explainWithCalcite( public void analyzeWithCalcite( String query, - List querySegments, UnresolvedPlan plan, QueryType queryType, // boolean disableCache, ResponseListener listener) { @@ -403,42 +400,7 @@ public void onFailure(Exception e) { 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 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) - .possibleCacheHit(possibleCacheHit) - .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( () -> { @@ -449,17 +411,14 @@ 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); AtomicReference physicalPlanRef = new AtomicReference<>(); - AtomicReference 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)); })) { @@ -483,16 +442,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 operatorTree = - buildOperatorTree( - querySegments, - logicalPlanNodes, - context.getNodeIdMappings(), - calcitePlan, - physicalRelRef.get(), - profile); - // Convert QueryResponse results to analyze format. List schema = new ArrayList<>(); if (queryResponse.getSchema() != null) { @@ -515,50 +464,14 @@ public void onFailure(Exception e) { .toArray(Object[]::new); } - // Extract scan metadata for recommendations #2 and #3. - org.apache.calcite.rel.metadata.RelMetadataQuery mq = - calcitePlan.getCluster().getMetadataQuery(); - long totalIndexDocs = getIndexDocCount(plan, context); - if (totalIndexDocs <= 0) { - totalIndexDocs = getScanBaseRowCount(calcitePlan, mq); - } - Set dateFieldNames = getDateFieldNames(plan, context); - boolean isTimeSeriesIndex = - !dateFieldNames.isEmpty() - || hasDateField(calcitePlan) - || hasDateFieldByName(logicalPlanNodes); - boolean hasDateRangeFilter = - logicalPlanNodes.stream() - .anyMatch( - n -> - n.contains("LogicalFilter") - && dateFieldNames.stream() - .anyMatch( - f -> - n.contains(f) - || n.contains("TIMESTAMP") - || n.contains("date("))) - || physicalPlanNodes.stream() - .anyMatch( - n -> - n.contains("FILTER") - && dateFieldNames.stream().anyMatch(n::contains)); - List recommendations = - buildRecommendations( - operatorTree, - profile, - totalIndexDocs, - isTimeSeriesIndex, - hasDateRangeFilter); + new AnalyzeRecommendationBuilder(profile).build(); AnalyzeResponse response = AnalyzeResponse.builder() - .query(query) - .querySegments(querySegments) + // .query(query) .logicalPlan(logicalPlanNodes) .physicalPlan(physicalPlanNodes) - .operator_tree(operatorTree) .recommendations(recommendations) .profile(profile) .possibleCacheHit(possibleCacheHit) @@ -581,676 +494,6 @@ public void onFailure(Exception e) { settings); } - private List buildOperatorTree( - List querySegments, - List logicalPlanNodes, - List nodeIdMappings, - RelNode logicalPlan, - RelNode physicalPlan, - QueryProfile profile) { - // Build a map from RelNode id to its logical plan description string. - Map 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> exclusiveIds = new ArrayList<>(); - Set previousIds = new HashSet<>(); - for (CalcitePlanContext.NodeIdMapping mapping : nodeIdMappings) { - Set current = new HashSet<>(mapping.relNodeIds()); - Set 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 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 idToRowCount = new HashMap<>(); - collectRowCounts(logicalPlan, mq, idToRowCount); - - // Collect non-cumulative costs per logical node for cost attribution. - Map idToCost = new HashMap<>(); - collectNonCumulativeCosts(logicalPlan, mq, idToCost); - - // 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 physicalTimings = new ArrayList<>(); - if (profile != null && profile.getPlan() != null) { - List 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}); - } - } - - // Collect per-segment plan IDs for cost attribution across all segments. - List> allSegmentPlanIds = new ArrayList<>(); - for (int i = 0; i < querySegments.size(); i++) { - Set ids = i < exclusiveIds.size() ? exclusiveIds.get(i) : Set.of(); - allSegmentPlanIds.add( - ids.stream() - .filter(idToDescription::containsKey) - .collect(java.util.stream.Collectors.toSet())); - } - List allSegmentCosts = computeSegmentCosts(allSegmentPlanIds, idToCost); - - List operators = new ArrayList<>(); - int physicalIdx = 0; - - // Build the pushed-down merged entry (first pushedSegments segments) - if (pushedSegments > 1) { - List mergedSegments = querySegments.subList(0, pushedSegments); - List descriptions = new ArrayList<>(); - for (int idx = 0; idx < pushedSegments; idx++) { - Set 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 nodeTypes = - mergedSegments.stream().map(AnalyzeResponse.QuerySegment::getNodeType).toList(); - List nodeCosts = allSegmentCosts.subList(0, pushedSegments); - // Collect all plan node ids in the pushed group for estimated_rows - Set allPushedPlanIds = new HashSet<>(); - for (int i = 0; i < pushedSegments; i++) { - Set 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) - .node_cost(nodeCosts) - .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 ids = !exclusiveIds.isEmpty() ? exclusiveIds.get(0) : Set.of(); - Set planIds = - ids.stream() - .filter(idToDescription::containsKey) - .collect(java.util.stream.Collectors.toSet()); - List 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())) - .node_cost(List.of(allSegmentCosts.get(0))) - .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 group = new ArrayList<>(); - List descriptions = new ArrayList<>(); - Set groupPlanIds = new HashSet<>(); - List groupCosts = new ArrayList<>(); - long logicalNodesInGroup = 0; - while (idx < querySegments.size() && logicalNodesInGroup < 1) { - group.add(querySegments.get(idx)); - groupCosts.add(allSegmentCosts.get(idx)); - Set 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 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) - .node_cost(groupCosts) - .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 inputs = current.getInputs(); - current = inputs.isEmpty() ? null : inputs.get(0); - } - return depth; - } - - private static RelNode getLeafScanNode(RelNode node) { - RelNode current = node; - while (current != null) { - List inputs = current.getInputs(); - if (inputs.isEmpty()) { - return current; - } - current = inputs.get(0); - } - return node; - } - - private static long getIndexDocCount(UnresolvedPlan plan, CalcitePlanContext context) { - try { - String[] indexNames = extractIndexNames(plan); - if (indexNames.length == 0) { - return -1; - } - org.apache.calcite.schema.SchemaPlus schema = context.config.getDefaultSchema(); - org.apache.calcite.schema.Table calciteTable = schema.getTable(indexNames[0]); - if (calciteTable instanceof org.opensearch.sql.storage.Table storageTable) { - return storageTable.getDocCount(); - } - } catch (Exception ignored) { - } - return -1; - } - - private static long getScanBaseRowCount( - RelNode plan, org.apache.calcite.rel.metadata.RelMetadataQuery mq) { - RelNode leaf = getLeafScanNode(plan); - try { - Double rowCount = mq.getRowCount(leaf); - if (rowCount != null) { - return Math.round(rowCount); - } - } catch (Exception ignored) { - } - return -1; - } - - private static boolean hasDateField(RelNode plan) { - RelNode leaf = getLeafScanNode(plan); - try { - org.apache.calcite.rel.type.RelDataType rowType; - if (leaf instanceof org.apache.calcite.rel.core.TableScan tableScan) { - rowType = tableScan.getTable().getRowType(); - } else { - rowType = leaf.getRowType(); - } - for (org.apache.calcite.rel.type.RelDataTypeField field : rowType.getFieldList()) { - org.apache.calcite.rel.type.RelDataType fieldType = field.getType(); - org.apache.calcite.sql.type.SqlTypeName typeName = fieldType.getSqlTypeName(); - if (typeName == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP - || typeName == org.apache.calcite.sql.type.SqlTypeName.DATE - || typeName == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE) { - return true; - } - if (fieldType - instanceof org.opensearch.sql.calcite.type.AbstractExprRelDataType exprType) { - org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT udt = exprType.getUdt(); - if (udt == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_TIMESTAMP - || udt == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_DATE) { - return true; - } - } - } - } catch (Exception ignored) { - } - return false; - } - - /** - * Checks whether the index backing the query has a date/timestamp field by looking up the full - * table schema from the Calcite schema registry. Unlike {@link #hasDateField(RelNode)}, this - * approach is not affected by project pushdown which narrows the scan's row type to only the - * fields referenced in the query. - */ - private static boolean hasDateFieldFromSchema(UnresolvedPlan plan, CalcitePlanContext context) { - try { - String[] indexNames = extractIndexNames(plan); - if (indexNames.length == 0) { - return false; - } - org.apache.calcite.schema.SchemaPlus schema = context.config.getDefaultSchema(); - for (String indexName : indexNames) { - org.apache.calcite.schema.Table table = schema.getTable(indexName); - if (table == null) { - continue; - } - org.apache.calcite.rel.type.RelDataType fullRowType = - table.getRowType(org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.TYPE_FACTORY); - for (org.apache.calcite.rel.type.RelDataTypeField field : fullRowType.getFieldList()) { - org.apache.calcite.rel.type.RelDataType fieldType = field.getType(); - org.apache.calcite.sql.type.SqlTypeName typeName = fieldType.getSqlTypeName(); - if (typeName == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP - || typeName == org.apache.calcite.sql.type.SqlTypeName.DATE - || typeName - == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE) { - return true; - } - if (fieldType - instanceof org.opensearch.sql.calcite.type.AbstractExprRelDataType exprType) { - org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT udt = exprType.getUdt(); - if (udt == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_TIMESTAMP - || udt - == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_DATE) { - return true; - } - } - } - } - } catch (Exception ignored) { - } - return false; - } - - private static Set getDateFieldNames(UnresolvedPlan plan, CalcitePlanContext context) { - Set dateFields = new HashSet<>(); - try { - String[] indexNames = extractIndexNames(plan); - if (indexNames.length == 0) { - return dateFields; - } - org.apache.calcite.schema.SchemaPlus schema = context.config.getDefaultSchema(); - for (String indexName : indexNames) { - org.apache.calcite.schema.Table table = schema.getTable(indexName); - if (table == null) { - continue; - } - org.apache.calcite.rel.type.RelDataType fullRowType = - table.getRowType(org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.TYPE_FACTORY); - for (org.apache.calcite.rel.type.RelDataTypeField field : fullRowType.getFieldList()) { - org.apache.calcite.rel.type.RelDataType fieldType = field.getType(); - org.apache.calcite.sql.type.SqlTypeName typeName = fieldType.getSqlTypeName(); - if (typeName == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP - || typeName == org.apache.calcite.sql.type.SqlTypeName.DATE - || typeName - == org.apache.calcite.sql.type.SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE) { - dateFields.add(field.getName()); - } else if (fieldType - instanceof org.opensearch.sql.calcite.type.AbstractExprRelDataType exprType) { - org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT udt = exprType.getUdt(); - if (udt == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_TIMESTAMP - || udt - == org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT.EXPR_DATE) { - dateFields.add(field.getName()); - } - } - } - } - } catch (Exception ignored) { - } - return dateFields; - } - - private static boolean hasDateFieldByName(List logicalPlanNodes) { - for (String node : logicalPlanNodes) { - if (node.contains("timestamp") - || node.contains("@timestamp") - || node.contains("event_time") - || node.contains("created_at") - || node.contains("updated_at") - || node.contains("date_field") - || node.contains("EXPR_TIMESTAMP") - || node.contains("EXPR_DATE")) { - return true; - } - } - return false; - } - - private void collectRowCounts( - RelNode node, - org.apache.calcite.rel.metadata.RelMetadataQuery mq, - Map 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 void collectNonCumulativeCosts( - RelNode node, - org.apache.calcite.rel.metadata.RelMetadataQuery mq, - Map idToCost) { - try { - org.apache.calcite.plan.RelOptCost cost = mq.getNonCumulativeCost(node); - if (cost != null && !cost.isInfinite()) { - double weight = cost.getCpu() + cost.getIo() * 10.0 + cost.getRows(); - idToCost.put(node.getId(), weight); - } - } catch (Exception ignored) { - } - for (RelNode input : node.getInputs()) { - collectNonCumulativeCosts(input, mq, idToCost); - } - } - - private Long getEstimatedRows(Set ids, Map idToRowCount) { - return ids.stream() - .filter(idToRowCount::containsKey) - .max(Integer::compareTo) - .map(id -> Math.round(idToRowCount.get(id))) - .orElse(null); - } - - /** - * Compute per-segment cost fractions from Calcite's non-cumulative cost. Each segment's cost is - * the sum of its exclusive RelNode costs, normalized to a percentage of the total across all - * segments in the operator tree. - */ - private List computeSegmentCosts( - List> segmentPlanIds, Map idToCost) { - List rawCosts = new ArrayList<>(); - for (Set ids : segmentPlanIds) { - double segCost = ids.stream().filter(idToCost::containsKey).mapToDouble(idToCost::get).sum(); - rawCosts.add(segCost); - } - double total = rawCosts.stream().mapToDouble(Double::doubleValue).sum(); - if (total <= 0) { - return rawCosts.stream().map(c -> 0f).toList(); - } - return rawCosts.stream().map(c -> (float) (c / total * 100.0)).toList(); - } - - private List buildRecommendations( - List operatorTree, - QueryProfile profile, - long totalIndexDocs, - boolean isTimeSeriesIndex, - boolean hasDateRangeFilter) { - List recommendations = new ArrayList<>(); - if (operatorTree == null || operatorTree.isEmpty() || profile == null) { - return recommendations; - } - - QueryProfile.Phase executePhase = profile.getPhases().get("execute"); - if (executePhase == null || executePhase.getTimeMillis() <= 0) { - return recommendations; - } - double executeTime = executePhase.getTimeMillis(); - - double maxTime = 0; - AnalyzeResponse.OperatorNode bottleneck = null; - - for (AnalyzeResponse.OperatorNode node : operatorTree) { - if (node.getActual_time_ms() == null) { - continue; - } - double time = parseTimeMs(node.getActual_time_ms()); - if (time > maxTime) { - maxTime = time; - bottleneck = node; - } - } - - int totalNodes = operatorTree.size(); - int pushedDown = 0; - for (AnalyzeResponse.OperatorNode node : operatorTree) { - if (Boolean.TRUE.equals(node.getIs_pushed_down())) { - pushedDown++; - } - } - int inMemory = totalNodes - pushedDown; - if (totalNodes > 0) { - recommendations.add( - AnalyzeResponse.Recommendation.builder() - .serverity(AnalyzeResponse.RecommendationSeverityLevel.INFO) - .rule("Pushdown visibility") - .message( - pushedDown - + " of " - + totalNodes - + " stages pushed down; " - + inMemory - + " ran in-memory") - .build()); - } - - if (bottleneck != null && maxTime > 0) { - long pct = Math.round((maxTime / executeTime) * 100); - String stage = - (bottleneck.getNode_type() != null && !bottleneck.getNode_type().isEmpty()) - ? String.join(", ", bottleneck.getNode_type()) - : "unknown"; - recommendations.add( - AnalyzeResponse.Recommendation.builder() - .serverity(AnalyzeResponse.RecommendationSeverityLevel.INFO) - .rule("Bottleneck stage") - .message(pct + "% of time is in the *" + stage + "* stage") - .affected_node(bottleneck.getSource()) - .suggestion("Consider optimizing the " + stage + " operation") - .build()); - } - - // In-memory bottleneck: find the non-pushed-down node with the highest self-time - double maxInMemoryTime = 0; - AnalyzeResponse.OperatorNode inMemoryBottleneck = null; - for (AnalyzeResponse.OperatorNode node : operatorTree) { - if (Boolean.TRUE.equals(node.getIs_pushed_down())) { - continue; - } - if (node.getActual_time_ms() == null) { - continue; - } - double time = parseTimeMs(node.getActual_time_ms()); - if (time > maxInMemoryTime) { - maxInMemoryTime = time; - inMemoryBottleneck = node; - } - } - if (inMemoryBottleneck != null - && maxInMemoryTime > 0 - && inMemoryBottleneck.getActual_rows() != null) { - long pct = Math.round((maxInMemoryTime / executeTime) * 100); - String stage = - (inMemoryBottleneck.getNode_type() != null - && !inMemoryBottleneck.getNode_type().isEmpty()) - ? String.join(", ", inMemoryBottleneck.getNode_type()) - : "unknown"; - recommendations.add( - AnalyzeResponse.Recommendation.builder() - .serverity(AnalyzeResponse.RecommendationSeverityLevel.WARNING) - .rule("In-memory bottleneck") - .message( - "Your *" - + stage - + "* ran in-memory over " - + inMemoryBottleneck.getActual_rows() - + " rows (" - + pct - + "% of time)") - .affected_node(inMemoryBottleneck.getSource()) - .suggestion( - "Consider pushing this operation down or reducing input rows with filters") - .build()); - } - - // Low scan selectivity: scan rows / total index docs > 80% - log.info( - "Low scan selectivity check: totalIndexDocs={}, operatorTree.size={}", - totalIndexDocs, - operatorTree.size()); - if (totalIndexDocs > 0) { - AnalyzeResponse.OperatorNode scanNode = operatorTree.get(0); - log.info( - "Low scan selectivity: scanNode.actual_rows={}, scanNode.estimated_rows={}", - scanNode.getActual_rows(), - scanNode.getEstimated_rows()); - if (scanNode.getActual_rows() != null && scanNode.getActual_rows() > 0) { - long scannedRows = scanNode.getActual_rows(); - long pct = Math.round((double) scannedRows / totalIndexDocs * 100); - long resultRows = - operatorTree.get(operatorTree.size() - 1).getActual_rows() != null - ? operatorTree.get(operatorTree.size() - 1).getActual_rows() - : 0; - log.info( - "Low scan selectivity: scannedRows={}, pct={}, resultRows={}", - scannedRows, - pct, - resultRows); - if (pct > 80) { - recommendations.add( - AnalyzeResponse.Recommendation.builder() - .serverity(AnalyzeResponse.RecommendationSeverityLevel.WARNING) - .rule("Low scan selectivity") - .message( - "Scanned " - + scannedRows - + " docs (" - + pct - + "% of index) to return " - + resultRows - + " rows") - .affected_node(scanNode.getSource()) - .suggestion("Add filters to reduce the number of documents scanned") - .build()); - } - } - } - - // Missing time filter: time-series index with no date range predicate pushed down - if (isTimeSeriesIndex && !hasDateRangeFilter) { - AnalyzeResponse.OperatorNode scanNode = operatorTree.get(0); - recommendations.add( - AnalyzeResponse.Recommendation.builder() - .serverity(AnalyzeResponse.RecommendationSeverityLevel.CRITICAL) - .rule("Missing time filter") - .message("No time filter on a time-series index: add one") - .affected_node(scanNode.getSource()) - .suggestion( - "Add a time range filter (e.g. where @timestamp > now() - interval 1 hour)") - .build()); - } - - return recommendations; - } - - private static double parseTimeMs(String timeMsStr) { - String stripped = timeMsStr.replaceAll("[^0-9.]", ""); - try { - return Double.parseDouble(stripped); - } catch (NumberFormatException e) { - return 0; - } - } - public void executeWithLegacy( UnresolvedPlan plan, QueryType queryType, 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. + * + *

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. + * + *

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}. + * + *

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 build() { + List 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 ineffectiveFilter() { + List 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 joinRowExplosion() { + List 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 expensiveSort() { + List 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 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 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 planNodes() { + List nodes = new ArrayList<>(); + if (profile.getPlan() instanceof PlanNode root) { + collect(root, nodes); + } + return nodes; + } + + private static void collect(PlanNode node, List 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 querySegments; private final UnresolvedPlan plan; private final QueryService queryService; private final ResponseListener listener; @@ -29,13 +26,11 @@ public AnalyzePlan( QueryId queryId, QueryType queryType, String query, - List querySegments, UnresolvedPlan plan, QueryService queryService, ResponseListener 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 querySegments, UnresolvedPlan plan, QueryType queryType, ResponseListener 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..aebe078fb85 --- /dev/null +++ b/core/src/test/java/org/opensearch/sql/executor/analyze/AnalyzeRecommendationBuilderTest.java @@ -0,0 +1,160 @@ +/* + * 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 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 ruleOf(List 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 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 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 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 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 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 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 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 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 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 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 recs = new AnalyzeRecommendationBuilder(profile(100, 200, scan)).build(); + assertTrue(ruleOf(recs, "Optimize Phase Dominates").isEmpty()); + } +} diff --git a/docs/user/ppl/interfaces/endpoint.md b/docs/user/ppl/interfaces/endpoint.md index 6704cc0cd48..662bf750dd1 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 @@ -171,11 +171,6 @@ 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", @@ -205,23 +200,6 @@ Expected output (trimmed): ] } }, - "operator_tree": [ - { - "source": "source=accounts | where age < 30", - "node_type": [ - "SearchFrom", - "WhereCommand" - ], - "description": [ - "CalciteLogicalIndexScan(table=[[OpenSearch, accounts]]): rowcount = 10000.0, cumulative cost = {99000.0 rows, 0.0 cpu, 0.0 io}, id = 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 - }, - ], "recommendations": [] } ``` @@ -230,11 +208,8 @@ 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. | @@ -242,26 +217,10 @@ Expected output (trimmed): | `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. | - - - ### 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. 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..6ef084a986d 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 @@ -312,39 +199,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/integ-test/src/yamlRestTest/resources/rest-api-spec/test/api/ppl.profile.yml b/integ-test/src/yamlRestTest/resources/rest-api-spec/test/api/ppl.profile.yml index d8bbb3d5724..55c913f8212 100644 --- a/integ-test/src/yamlRestTest/resources/rest-api-spec/test/api/ppl.profile.yml +++ b/integ-test/src/yamlRestTest/resources/rest-api-spec/test/api/ppl.profile.yml @@ -56,11 +56,6 @@ teardown: - gte: {profile.phases.format.time_ms: 0.0} - gt: {profile.plan.time_ms: 0.0} - match: {profile.plan.rows: 2} - - match: {query: 'source=ppl_profile | fields message'} - - is_true: logicalPlan - - is_true: physicalPlan - - is_true: operator_tree - - is_true: recommendations - is_true: schema - is_true: datarows - match: {total: 2} diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java b/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java index 772f1ec123f..7af0a5981b4 100644 --- a/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java +++ b/plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java @@ -222,7 +222,7 @@ protected void doExecute( * Removing `|| transformedRequest.profile()` from line 200 will separate the `profile` and * `analyze` endpoints. See PR #5568. */ - } else if (transformedRequest.analyze() || transformedRequest.profile()) { + } else if (transformedRequest.analyze()) { // || transformedRequest.profile()) { pplService.analyze( transformedRequest, createAnalyzeResponseListener(transformedRequest, clearingListener)); } else { 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 l QueryContext.getRequestId(), anonymizer.anonymizeStatement(statement)); - List 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 extractQuerySegments(ParseTree cst, String queryText) { - List 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 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 mappings = context.getNodeIdMappings(); - assertFalse(mappings.isEmpty()); - - List 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 mappings = context.getNodeIdMappings(); - List 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 mappings = context.getNodeIdMappings(); - List 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 mappings = context.getNodeIdMappings(); - List 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 mappings = context.getNodeIdMappings(); - List 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 mappings = context.getNodeIdMappings(); - List 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 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 mappings = context.getNodeIdMappings(); - List 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")); - } -} From 3954423a35ca9b30b08142299c7b750e746e7e22 Mon Sep 17 00:00:00 2001 From: Krish Gandhi Date: Thu, 13 Aug 2026 15:52:44 -0700 Subject: [PATCH 3/4] Adding tests, fixing tests Signed-off-by: Krish Gandhi --- .../opensearch/sql/executor/QueryService.java | 76 +++++++++++- .../AnalyzeRecommendationBuilderTest.java | 108 ++++++++++++++++++ .../sql/calcite/remote/CalciteAnalyzeIT.java | 77 +++++++++++++ 3 files changed, 258 insertions(+), 3 deletions(-) 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 0b7a03629d7..cabc653015a 100644 --- a/core/src/main/java/org/opensearch/sql/executor/QueryService.java +++ b/core/src/main/java/org/opensearch/sql/executor/QueryService.java @@ -21,13 +21,19 @@ import org.apache.calcite.plan.RelTraitDef; import org.apache.calcite.rel.RelCollation; import org.apache.calcite.rel.RelCollations; +import org.apache.calcite.rel.RelHomogeneousShuttle; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.RelRoot; import org.apache.calcite.rel.core.Sort; import org.apache.calcite.rel.logical.LogicalSort; +import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexShuttle; import org.apache.calcite.runtime.Hook; import org.apache.calcite.schema.SchemaPlus; import org.apache.calcite.sql.SqlExplainLevel; +import org.apache.calcite.sql.SqlOperator; +import org.apache.calcite.sql.fun.SqlStdOperatorTable; import org.apache.calcite.sql.parser.SqlParser; import org.apache.calcite.tools.FrameworkConfig; import org.apache.calcite.tools.Frameworks; @@ -218,7 +224,9 @@ public void executeWithCalcite( RelNode calcitePlan = StageErrorHandler.executeStage( QueryProcessingStage.PLAN_CONVERSION, - () -> convertToCalcitePlan(relNode, context), + () -> + withCheckedArithmetic( + convertToCalcitePlan(relNode, context), context), "while converting the query to an executable plan"); executeCalcitePlan(calcitePlan, context, listener, analyzeMetric, analyzeStart); @@ -293,7 +301,8 @@ public void explainWithCalcite( context.run( () -> { RelNode relNode = analyze(plan, context); - RelNode calcitePlan = convertToCalcitePlan(relNode, context); + RelNode calcitePlan = + withCheckedArithmetic(convertToCalcitePlan(relNode, context), context); if (format != null) { executionEngine.explain(calcitePlan, mode, format, context, listener); } else { @@ -412,7 +421,8 @@ public void onFailure(Exception e) { CalcitePlanContext.create( buildFrameworkConfig(), SysLimit.fromSettings(settings), queryType); RelNode relNode = analyze(plan, context); - RelNode calcitePlan = convertToCalcitePlan(relNode, context); + RelNode calcitePlan = + withCheckedArithmetic(convertToCalcitePlan(relNode, context), context); AtomicReference physicalPlanRef = new AtomicReference<>(); try (Hook.Closeable closeable = @@ -628,6 +638,66 @@ private boolean isCalciteEnabled(Settings settings) { } } + /** + * Rewrite {@code +}/{@code -}/{@code *} to their overflow-checked variants ({@code CHECKED_PLUS} + * / {@code CHECKED_MINUS} / {@code CHECKED_MULTIPLY}) so integer and long arithmetic overflow + * throws {@link ArithmeticException} (via {@code Math.addExact} etc.) instead of silently + * wrapping. Applied before pushdown so both coordinator-executed and pushed-down (script) + * arithmetic are checked. Floating-point arithmetic is unchanged (IEEE 754). + * + *

This does the same rewrite as Calcite's {@code ConvertToChecked} but preserves each call's + * originally inferred type (via {@code makeCall(type, op, operands)}) and touches only the three + * arithmetic operators, so it does not re-derive the types of unrelated calls (e.g. {@code + * CEIL}/{@code DIVIDE}) the way {@code ConvertToChecked} does. + */ + private static RelNode withCheckedArithmetic(RelNode calcitePlan, CalcitePlanContext context) { + RexShuttle checkedShuttle = + new RexShuttle() { + @Override + public RexNode visitCall(RexCall call) { + RexNode visited = super.visitCall(call); + if (!(visited instanceof RexCall rexCall)) { + return visited; + } + SqlOperator checked = + switch (rexCall.getOperator().getKind()) { + case PLUS -> SqlStdOperatorTable.CHECKED_PLUS; + case MINUS -> SqlStdOperatorTable.CHECKED_MINUS; + case TIMES -> SqlStdOperatorTable.CHECKED_MULTIPLY; + default -> null; + }; + // Only integer/long arithmetic can overflow silently and has a checked + // implementation (Math.addExact etc.). Float/double/decimal have no checked variant + // (SqlFunctions.checkedMultiply(double,double) does not exist) and follow IEEE 754, so + // leave them untouched. + if (checked == null || !isCheckableIntegerArithmetic(rexCall)) { + return visited; + } + return context.rexBuilder.makeCall(rexCall.getType(), checked, rexCall.getOperands()); + } + }; + return calcitePlan.accept( + new RelHomogeneousShuttle() { + @Override + public RelNode visit(RelNode other) { + RelNode visited = super.visitChildren(other); + return visited.accept(checkedShuttle); + } + }); + } + + /** Returns whether the result and every operand are BIGINT. */ + private static boolean isCheckableIntegerArithmetic(RexCall call) { + if (!isCheckableLongType(call.getType())) { + return false; + } + return call.getOperands().stream().allMatch(op -> isCheckableLongType(op.getType())); + } + + private static boolean isCheckableLongType(org.apache.calcite.rel.type.RelDataType type) { + return type.getSqlTypeName() == org.apache.calcite.sql.type.SqlTypeName.BIGINT; + } + /** * Walk the cause chain to find an {@link ArithmeticException} raised by checked arithmetic. Row- * level overflow surfaces wrapped (SQLException -> RuntimeException -> ErrorReport), so a 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 index aebe078fb85..29974906e4e 100644 --- a/core/src/test/java/org/opensearch/sql/executor/analyze/AnalyzeRecommendationBuilderTest.java +++ b/core/src/test/java/org/opensearch/sql/executor/analyze/AnalyzeRecommendationBuilderTest.java @@ -157,4 +157,112 @@ void optimizePhaseDominatesSilentWhenExecutionLarger() { List 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 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 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 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 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 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 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 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 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 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 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/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 6ef084a986d..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 @@ -166,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 @@ -180,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 From 9fcb5e716f1a88ee1ae6bc27cf6731381923393f Mon Sep 17 00:00:00 2001 From: Krish Gandhi Date: Fri, 14 Aug 2026 09:37:42 -0700 Subject: [PATCH 4/4] Fixing CI, cleaning up unused code Signed-off-by: Krish Gandhi --- .../java/org/opensearch/sql/storage/Table.java | 9 --------- .../sql/opensearch/client/OpenSearchClient.java | 8 -------- .../opensearch/client/OpenSearchNodeClient.java | 17 ----------------- .../opensearch/client/OpenSearchRestClient.java | 13 ------------- .../sql/opensearch/storage/OpenSearchIndex.java | 5 ----- 5 files changed, 52 deletions(-) diff --git a/core/src/main/java/org/opensearch/sql/storage/Table.java b/core/src/main/java/org/opensearch/sql/storage/Table.java index 13f3e86ff4b..33dbd7d66d3 100644 --- a/core/src/main/java/org/opensearch/sql/storage/Table.java +++ b/core/src/main/java/org/opensearch/sql/storage/Table.java @@ -35,15 +35,6 @@ default void create(Map schema) { throw new UnsupportedOperationException("Unsupported Operation"); } - /** - * Get the total document count for this table. - * - * @return total document count, or -1 if unavailable - */ - default long getDocCount() { - return -1; - } - /** Get the {@link ExprType} for each field in the table. */ Map getFieldTypes(); diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchClient.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchClient.java index a9d5b25c14f..68350c5a0fd 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchClient.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchClient.java @@ -55,14 +55,6 @@ public interface OpenSearchClient { */ Map getIndexMaxResultWindows(String... indexExpression); - /** - * Get the total document count for the given index expression. - * - * @param indexExpression index expression - * @return total document count - */ - long getIndexDocCount(String indexExpression); - /** * Perform search query in the search request. * 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 0f76e981fe8..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,23 +126,6 @@ public Map 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 long getIndexDocCount(String indexExpression) { - try { - org.opensearch.action.admin.indices.stats.IndicesStatsResponse response = - client.admin().indices().prepareStats(indexExpression).clear().setDocs(true).get(); - return response.getTotal().getDocs().getCount(); - } catch (Exception e) { - return -1; - } - } - @Override public Map getIndexMaxResultWindows(String... indexExpression) { try { diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchRestClient.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchRestClient.java index e841ad1a739..f369c0003b8 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchRestClient.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchRestClient.java @@ -88,19 +88,6 @@ public Map getIndexMappings(String... indexExpression) { } } - @Override - public long getIndexDocCount(String indexExpression) { - try { - org.opensearch.client.core.CountRequest countRequest = - new org.opensearch.client.core.CountRequest(indexExpression); - org.opensearch.client.core.CountResponse response = - client.count(countRequest, RequestOptions.DEFAULT); - return response.getCount(); - } catch (Exception e) { - return -1; - } - } - @Override public Map getIndexMaxResultWindows(String... indexExpression) { GetSettingsRequest request = diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java index 1ab960da4e0..3350c00fb0c 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java @@ -191,11 +191,6 @@ public Map getFieldOpenSearchTypes() { return cachedFieldOpenSearchTypes; } - @Override - public long getDocCount() { - return client.getIndexDocCount(indexName.toString()); - } - /** Get the max result window setting of the table. */ public Integer getMaxResultWindow() { if (cachedMaxResultWindow == null) {