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 b715fdac86d..24f12454594 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 123fe7a10fe..c441552452e 100644 --- a/core/src/main/java/org/opensearch/sql/executor/QueryService.java +++ b/core/src/main/java/org/opensearch/sql/executor/QueryService.java @@ -44,8 +44,10 @@ import org.apache.calcite.tools.Programs; import org.opensearch.sql.analysis.AnalysisContext; import org.opensearch.sql.analysis.Analyzer; +import org.opensearch.sql.ast.Node; import org.opensearch.sql.ast.statement.ExplainMode; import org.opensearch.sql.ast.tree.HighlightConfig; +import org.opensearch.sql.ast.tree.Relation; import org.opensearch.sql.ast.tree.UnresolvedPlan; import org.opensearch.sql.calcite.CalcitePlanContext; import org.opensearch.sql.calcite.CalciteRelNodeVisitor; @@ -194,14 +196,16 @@ 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"); analyzeMetric.set(System.nanoTime() - analyzeStart); - executeCalcitePlan(calcitePlan, context, listener); + // Wrap execution with EXECUTING stage tracking + StageErrorHandler.executeStageVoid( + QueryProcessingStage.EXECUTING, + () -> executionEngine.execute(calcitePlan, context, listener), + "while running the query"); }, QueryService.class); } catch (Throwable t) { @@ -216,25 +220,6 @@ public void executeWithCalcite( settings); } - private void executeCalcitePlan( - RelNode calcitePlan, - CalcitePlanContext context, - ResponseListener listener) { - try { - StageErrorHandler.executeStageVoid( - QueryProcessingStage.EXECUTING, - () -> executionEngine.execute(calcitePlan, context, listener), - "while running the query"); - } catch (RuntimeException e) { - ArithmeticException overflow = findArithmeticOverflow(e); - if (overflow != null) { - throw new NonFallbackCalciteException( - "Arithmetic overflow: " + overflow.getMessage(), overflow); - } - throw e; - } - } - public void explainWithCalcite( UnresolvedPlan plan, QueryType queryType, @@ -264,8 +249,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 { @@ -291,7 +275,7 @@ public void analyzeWithCalcite( String query, List querySegments, UnresolvedPlan plan, - QueryType queryType, + QueryType queryType, // boolean disableCache, ResponseListener listener) { if (!shouldUseCalcite(queryType)) { listener.onFailure( @@ -300,10 +284,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<>(); @@ -342,8 +335,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) { @@ -351,6 +347,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(); @@ -378,8 +381,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) @@ -465,6 +469,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) @@ -472,8 +513,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) @@ -569,6 +611,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<>(); @@ -593,6 +639,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; @@ -615,6 +672,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++) { @@ -628,6 +686,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)) @@ -650,6 +709,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) @@ -666,9 +726,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() @@ -693,6 +755,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) @@ -728,6 +791,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, @@ -744,6 +979,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) @@ -752,6 +1004,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, @@ -886,82 +1339,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 - * top-level {@code catch (ArithmeticException)} is insufficient. - */ - private static ArithmeticException findArithmeticOverflow(@Nullable Throwable t) { - for (Throwable cause = t; - cause != null && cause != cause.getCause(); - cause = cause.getCause()) { - if (cause instanceof ArithmeticException arithmeticException) { - return arithmeticException; - } - } - return null; - } - // TODO https://github.com/opensearch-project/sql/issues/3457 // Calcite is not available for SQL query now. Maybe release in 3.1.0? private boolean shouldUseCalcite(QueryType queryType) { @@ -984,6 +1361,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 b491f38ef80..f7634efd353 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 @@ -129,6 +129,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 2e37782b72b..9e54aea4087 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 @@ -358,6 +358,26 @@ 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..95ea36d57cf 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,9 @@ 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 +144,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 +190,7 @@ public OpenSearchQueryRequest(StreamInput in, OpenSearchStorageEngine engine) th exprValueFactory = new OpenSearchExprValueFactory( index.getFieldOpenSearchTypes(), index.isFieldTypeTolerance()); + this.disableRequestCache = false; } @Override @@ -218,6 +224,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 +286,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) {