Improve the performance of TermsAggregation (#8959)

docs/en/changes/changes.md

@@ -39,6 +39,8 @@
 * Add the support eBPF profiling field into the process entity.
 * E2E: fix log test miss verify LAL and metrics.
 * Enhance Converter mechanism in kernel level to make BanyanDB native feature more effective.
+* Add TermsAggregation properties collect_mode and execution_hint.
+* Add "execution_hint": "map", "collect_mode": "breadth_first" for aggregation and topology query to improve 5-10x performance.
 
 #### UI
 

oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/aggregation/TermsAggregation.java

@@ -36,6 +36,8 @@ public final class TermsAggregation extends Aggregation {
     private final BucketOrder order;
     private final Integer size;
     private final ImmutableMap<String, Aggregation> aggregations;
+    private final TermsAggregationBuilder.CollectMode collectMode;
+    private final TermsAggregationBuilder.ExecutionHint executionHint;
 
     @Override
     public String name() {
@@ -58,6 +60,12 @@ public final class TermsAggregation extends Aggregation {
                     if (value.getOrder() != null) {
                         writeOrder(value, gen);
                     }
+                    if (value.getCollectMode() != null) {
+                        gen.writeStringField("collect_mode", value.getCollectMode().value);
+                    }
+                    if (value.getExecutionHint() != null) {
+                        gen.writeStringField("execution_hint", value.getExecutionHint().value);
+                    }
                 }
                 gen.writeEndObject();
 

oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/aggregation/TermsAggregationBuilder.java

@@ -31,6 +31,9 @@ public final class TermsAggregationBuilder implements AggregationBuilder {
     private Integer size;
     private ImmutableMap.Builder<String, Aggregation> subAggregations;
 
+    private CollectMode collectMode;
+    private ExecutionHint executionHint;
+
     TermsAggregationBuilder(final String name) {
         checkArgument(!Strings.isNullOrEmpty(name), "name cannot be blank");
         this.name = name;
@@ -65,6 +68,18 @@ public final class TermsAggregationBuilder implements AggregationBuilder {
         return subAggregation(subAggregationBuilder.build());
     }
 
+    public TermsAggregationBuilder collectMode(CollectMode collectMode) {
+        requireNonNull(collectMode, "collectMode");
+        this.collectMode = collectMode;
+        return this;
+    }
+
+    public TermsAggregationBuilder executionHint(ExecutionHint executionHint) {
+        requireNonNull(executionHint, "executionHint");
+        this.executionHint = executionHint;
+        return this;
+    }
+
     @Override
     public TermsAggregation build() {
         ImmutableMap<String, Aggregation> subAggregations;
@@ -74,7 +89,7 @@ public final class TermsAggregationBuilder implements AggregationBuilder {
             subAggregations = this.subAggregations.build();
         }
         return new TermsAggregation(
-            name, field, order, size, subAggregations
+            name, field, order, size, subAggregations, collectMode, executionHint
         );
     }
 
@@ -84,4 +99,34 @@ public final class TermsAggregationBuilder implements AggregationBuilder {
         }
         return subAggregations;
     }
+
+    public enum CollectMode {
+        BREADTH_FIRST("breadth_first"), DEPTH_FIRST("depth_first");
+
+        final String value;
+
+        CollectMode(final String value) {
+            this.value = value;
+        }
+
+        @Override
+        public String toString() {
+            return value;
+        }
+    }
+
+    public enum ExecutionHint {
+        GLOBAL_ORDINALS("global_ordinals"), MAP("map");
+
+        final String value;
+
+        ExecutionHint(final String value) {
+            this.value = value;
+        }
+
+        @Override
+        public String toString() {
+            return value;
+        }
+    }
 }

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAO.java

@@ -28,6 +28,7 @@ import org.apache.skywalking.library.elasticsearch.requests.search.Search;
 import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
 import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.Aggregation;
 import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.BucketOrder;
+import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.TermsAggregationBuilder;
 import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.query.enumeration.Order;
@@ -105,6 +106,8 @@ public class AggregationQueryEsDAO extends EsDAO implements IAggregationQueryDAO
                        .order(BucketOrder.aggregation(valueColumnName, asc))
                        .size(condition.getTopN())
                        .subAggregation(Aggregation.avg(valueColumnName).field(valueColumnName))
+                       .executionHint(TermsAggregationBuilder.ExecutionHint.MAP)
+                       .collectMode(TermsAggregationBuilder.CollectMode.BREADTH_FIRST)
                        .build());
 
         final SearchResponse response = getClient().search(tableName, search.build());

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetricsQueryEsDAO.java

@@ -70,6 +70,8 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
         final TermsAggregationBuilder entityIdAggregation =
             Aggregation.terms(Metrics.ENTITY_ID)
                        .field(Metrics.ENTITY_ID)
+                       .executionHint(TermsAggregationBuilder.ExecutionHint.MAP)
+                       .collectMode(TermsAggregationBuilder.CollectMode.BREADTH_FIRST)
                        .size(1);
         functionAggregation(function, entityIdAggregation, valueColumnName);
 

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TopologyQueryEsDAO.java

@@ -26,6 +26,7 @@ import org.apache.skywalking.library.elasticsearch.requests.search.Query;
 import org.apache.skywalking.library.elasticsearch.requests.search.Search;
 import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
 import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.Aggregation;
+import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.TermsAggregationBuilder;
 import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
 import org.apache.skywalking.oap.server.core.UnexpectedException;
 import org.apache.skywalking.oap.server.core.analysis.manual.relation.endpoint.EndpointRelationServerSideMetrics;
@@ -206,7 +207,11 @@ public class TopologyQueryEsDAO extends EsDAO implements ITopologyQueryDAO {
                 .terms(Metrics.ENTITY_ID).field(Metrics.ENTITY_ID)
                 .subAggregation(
                     Aggregation.terms(ServiceRelationServerSideMetrics.COMPONENT_ID)
-                               .field(ServiceRelationServerSideMetrics.COMPONENT_ID))
+                               .field(ServiceRelationServerSideMetrics.COMPONENT_ID)
+                               .executionHint(TermsAggregationBuilder.ExecutionHint.MAP)
+                               .collectMode(TermsAggregationBuilder.CollectMode.BREADTH_FIRST))
+                .executionHint(TermsAggregationBuilder.ExecutionHint.MAP)
+                .collectMode(TermsAggregationBuilder.CollectMode.BREADTH_FIRST)
                 .size(1000));
 
         final String index =
@@ -242,8 +247,11 @@ public class TopologyQueryEsDAO extends EsDAO implements ITopologyQueryDAO {
                 .terms(Metrics.ENTITY_ID).field(Metrics.ENTITY_ID)
                 .subAggregation(
                     Aggregation.terms(ServiceInstanceRelationServerSideMetrics.COMPONENT_ID)
-                               .field(
-                                   ServiceInstanceRelationServerSideMetrics.COMPONENT_ID))
+                               .field(ServiceInstanceRelationServerSideMetrics.COMPONENT_ID)
+                               .executionHint(TermsAggregationBuilder.ExecutionHint.MAP)
+                               .collectMode(TermsAggregationBuilder.CollectMode.BREADTH_FIRST))
+                .executionHint(TermsAggregationBuilder.ExecutionHint.MAP)
+                .collectMode(TermsAggregationBuilder.CollectMode.BREADTH_FIRST)
                 .size(1000));
 
         final String index =
@@ -273,7 +281,10 @@ public class TopologyQueryEsDAO extends EsDAO implements ITopologyQueryDAO {
     private List<Call.CallDetail> loadEndpoint(SearchBuilder sourceBuilder, String indexName,
                                                DetectPoint detectPoint) {
         sourceBuilder.aggregation(
-            Aggregation.terms(Metrics.ENTITY_ID).field(Metrics.ENTITY_ID).size(1000));
+            Aggregation.terms(Metrics.ENTITY_ID).field(Metrics.ENTITY_ID)
+                       .executionHint(TermsAggregationBuilder.ExecutionHint.MAP)
+                       .collectMode(TermsAggregationBuilder.CollectMode.BREADTH_FIRST)
+                       .size(1000));
 
         final String index =
             IndexController.LogicIndicesRegister.getPhysicalTableName(indexName);