Support Dynamic Pruning in Cardinality Aggregation (opensearch-projec…

…t#13821) * Cardinality aggregation dynamic pruning changes Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * Reading Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * remaining disjunction scorer full understand Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * utilize competitive iterator api to perform pruning Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * handle missing input Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * add change log Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * clean up Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * Clean up Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * Test fix Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * Do all the scoring within Cardinality Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * clean unnecessary Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * fix Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * Add dynamic flag for this feature Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * Add random test, small bug fix Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * address comment Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * Address comments Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> * address comments Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> --------- Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com> Co-authored-by: Rishabh Maurya <rishabhmaurya05@gmail.com>
harshavamsi · Jul 12, 2024 · b59afbb · b59afbb
1 parent 6d81890
commit b59afbb
Show file tree

Hide file tree

Showing 9 changed files with 623 additions and 11 deletions.
diff --git a/...rc/internalClusterTest/java/org/opensearch/search/aggregations/metrics/CardinalityIT.java b/...rc/internalClusterTest/java/org/opensearch/search/aggregations/metrics/CardinalityIT.java
@@ -34,6 +34,7 @@
 
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
+import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse;
 import org.opensearch.action.index.IndexRequestBuilder;
 import org.opensearch.action.search.SearchResponse;
 import org.opensearch.common.settings.Settings;
@@ -59,6 +60,7 @@
 import static java.util.Collections.emptyMap;
 import static org.opensearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.opensearch.index.query.QueryBuilders.matchAllQuery;
+import static org.opensearch.search.SearchService.CARDINALITY_AGGREGATION_PRUNING_THRESHOLD;
 import static org.opensearch.search.SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING;
 import static org.opensearch.search.aggregations.AggregationBuilders.cardinality;
 import static org.opensearch.search.aggregations.AggregationBuilders.global;
@@ -255,6 +257,36 @@ public void testSingleValuedString() throws Exception {
         assertCount(count, numDocs);
     }
 
+    public void testDisableDynamicPruning() throws Exception {
+        SearchResponse response = client().prepareSearch("idx")
+            .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_value"))
+            .get();
+        assertSearchResponse(response);
+
+        Cardinality count1 = response.getAggregations().get("cardinality");
+
+        final ClusterUpdateSettingsResponse updateSettingResponse = client().admin()
+            .cluster()
+            .prepareUpdateSettings()
+            .setTransientSettings(Settings.builder().put(CARDINALITY_AGGREGATION_PRUNING_THRESHOLD.getKey(), 0))
+            .get();
+        assertEquals(updateSettingResponse.getTransientSettings().get(CARDINALITY_AGGREGATION_PRUNING_THRESHOLD.getKey()), "0");
+
+        response = client().prepareSearch("idx")
+            .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_value"))
+            .get();
+        assertSearchResponse(response);
+        Cardinality count2 = response.getAggregations().get("cardinality");
+
+        assertEquals(count1, count2);
+
+        client().admin()
+            .cluster()
+            .prepareUpdateSettings()
+            .setTransientSettings(Settings.builder().putNull(CARDINALITY_AGGREGATION_PRUNING_THRESHOLD.getKey()))
+            .get();
+    }
+
     public void testSingleValuedNumeric() throws Exception {
         SearchResponse response = client().prepareSearch("idx")
             .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field(singleNumericField()))

diff --git a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java
@@ -540,6 +540,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                 SearchService.MAX_OPEN_PIT_CONTEXT,
                 SearchService.MAX_PIT_KEEPALIVE_SETTING,
                 SearchService.MAX_AGGREGATION_REWRITE_FILTERS,
+                SearchService.CARDINALITY_AGGREGATION_PRUNING_THRESHOLD,
                 CreatePitController.PIT_INIT_KEEP_ALIVE,
                 Node.WRITE_PORTS_FILE_SETTING,
                 Node.NODE_NAME_SETTING,

diff --git a/server/src/main/java/org/opensearch/search/DefaultSearchContext.java b/server/src/main/java/org/opensearch/search/DefaultSearchContext.java
@@ -106,6 +106,7 @@
 import java.util.function.Function;
 import java.util.function.LongSupplier;
 
+import static org.opensearch.search.SearchService.CARDINALITY_AGGREGATION_PRUNING_THRESHOLD;
 import static org.opensearch.search.SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING;
 import static org.opensearch.search.SearchService.MAX_AGGREGATION_REWRITE_FILTERS;
 
@@ -189,6 +190,7 @@ final class DefaultSearchContext extends SearchContext {
     private final boolean concurrentSearchSettingsEnabled;
     private final SetOnce<Boolean> requestShouldUseConcurrentSearch = new SetOnce<>();
     private final int maxAggRewriteFilters;
+    private final int cardinalityAggregationPruningThreshold;
 
     DefaultSearchContext(
         ReaderContext readerContext,
@@ -244,6 +246,7 @@ final class DefaultSearchContext extends SearchContext {
         this.requestToAggReduceContextBuilder = requestToAggReduceContextBuilder;
 
         this.maxAggRewriteFilters = evaluateFilterRewriteSetting();
+        this.cardinalityAggregationPruningThreshold = evaluateCardinalityAggregationPruningThreshold();
     }
 
     @Override
@@ -1010,4 +1013,16 @@ private int evaluateFilterRewriteSetting() {
         }
         return 0;
     }
+
+    @Override
+    public int cardinalityAggregationPruningThreshold() {
+        return cardinalityAggregationPruningThreshold;
+    }
+
+    private int evaluateCardinalityAggregationPruningThreshold() {
+        if (clusterService != null) {
+            return clusterService.getClusterSettings().get(CARDINALITY_AGGREGATION_PRUNING_THRESHOLD);
+        }
+        return 0;
+    }
 }
diff --git a/server/src/main/java/org/opensearch/search/SearchService.java b/server/src/main/java/org/opensearch/search/SearchService.java
@@ -288,6 +288,15 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
         Property.NodeScope
     );
 
+    // value 0 can disable dynamic pruning optimization in cardinality aggregation
+    public static final Setting<Integer> CARDINALITY_AGGREGATION_PRUNING_THRESHOLD = Setting.intSetting(
+        "search.dynamic_pruning.cardinality_aggregation.max_allowed_cardinality",
+        100,
+        0,
+        Property.Dynamic,
+        Property.NodeScope
+    );
+
     public static final int DEFAULT_SIZE = 10;
     public static final int DEFAULT_FROM = 0;