TopN selection for streaming terms aggregations

CHANGELOG.md

@@ -22,6 +22,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Introduce concurrent translog recovery to accelerate segment replication primary promotion ([#20251](https://github.com/opensearch-project/OpenSearch/pull/20251))
 - Update to `almalinux:10` ([#20482](https://github.com/opensearch-project/OpenSearch/pull/20482))
 - Add X-Request-Id to uniquely identify a search request ([#19798](https://github.com/opensearch-project/OpenSearch/pull/19798))
+- Added TopN selection logic for streaming terms aggregations ([#20481](https://github.com/opensearch-project/OpenSearch/pull/20481))
 
 ### Changed
 - Handle custom metadata files in subdirectory-store ([#20157](https://github.com/opensearch-project/OpenSearch/pull/20157))

server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java

@@ -171,6 +171,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 IndexSettings.MAX_SLICES_PER_SCROLL,
                 IndexSettings.MAX_SLICES_PER_PIT,
                 IndexSettings.MAX_REGEX_LENGTH_SETTING,
+                IndexSettings.STREAMING_AGGREGATION_MIN_SHARD_SIZE_SETTING,
                 ShardsLimitAllocationDecider.INDEX_TOTAL_SHARDS_PER_NODE_SETTING,
                 ShardsLimitAllocationDecider.INDEX_TOTAL_PRIMARY_SHARDS_PER_NODE_SETTING,
                 ShardsLimitAllocationDecider.INDEX_TOTAL_REMOTE_CAPABLE_SHARDS_PER_NODE_SETTING,

server/src/main/java/org/opensearch/index/IndexSettings.java

@@ -633,6 +633,19 @@ public static IndexMergePolicy fromString(String text) {
         Property.IndexScope
     );
 
+    /**
+     * Minimum shard size for streaming aggregations to ensure accuracy.
+     * This applies per-segment in streaming mode, not per-shard.
+     * Default is 1000. Can be adjusted based on accuracy requirements.
+     */
+    public static final Setting<Integer> STREAMING_AGGREGATION_MIN_SHARD_SIZE_SETTING = Setting.intSetting(
+        "index.aggregation.streaming.min_shard_size",
+        1000,
+        1,
+        Property.Dynamic,
+        Property.IndexScope
+    );
+
     public static final Setting<String> DEFAULT_PIPELINE = new Setting<>(
         "index.default_pipeline",
         IngestService.NOOP_PIPELINE_NAME,
@@ -998,6 +1011,10 @@ private void setRetentionLeaseMillis(final TimeValue retentionLease) {
      * The maximum length of regex string allowed in a regexp query.
      */
     private volatile int maxRegexLength;
+    /**
+     * The minimum shard size for streaming aggregations.
+     */
+    private volatile int streamingAggregationMinShardSize;
 
     /**
      * The max amount of time to wait for merges
@@ -1167,6 +1184,7 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         maxTermsCount = scopedSettings.get(MAX_TERMS_COUNT_SETTING);
         maxNestedQueryDepth = scopedSettings.get(MAX_NESTED_QUERY_DEPTH_SETTING);
         maxRegexLength = scopedSettings.get(MAX_REGEX_LENGTH_SETTING);
+        streamingAggregationMinShardSize = scopedSettings.get(STREAMING_AGGREGATION_MIN_SHARD_SIZE_SETTING);
         this.tieredMergePolicyProvider = new TieredMergePolicyProvider(logger, this);
         this.logByteSizeMergePolicyProvider = new LogByteSizeMergePolicyProvider(logger, this);
         this.indexSortConfig = new IndexSortConfig(this);
@@ -1299,6 +1317,7 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         scopedSettings.addSettingsUpdateConsumer(DEFAULT_FIELD_SETTING, this::setDefaultFields);
         scopedSettings.addSettingsUpdateConsumer(INDEX_SEARCH_IDLE_AFTER, this::setSearchIdleAfter);
         scopedSettings.addSettingsUpdateConsumer(MAX_REGEX_LENGTH_SETTING, this::setMaxRegexLength);
+        scopedSettings.addSettingsUpdateConsumer(STREAMING_AGGREGATION_MIN_SHARD_SIZE_SETTING, this::setStreamingAggregationMinShardSize);
         scopedSettings.addSettingsUpdateConsumer(DEFAULT_PIPELINE, this::setDefaultPipeline);
         scopedSettings.addSettingsUpdateConsumer(FINAL_PIPELINE, this::setRequiredPipeline);
         scopedSettings.addSettingsUpdateConsumer(INDEX_SOFT_DELETES_RETENTION_OPERATIONS_SETTING, this::setSoftDeleteRetentionOperations);
@@ -2052,6 +2071,17 @@ private void setMaxRegexLength(int maxRegexLength) {
         this.maxRegexLength = maxRegexLength;
     }
 
+    /**
+     * Returns the minimum shard size for streaming aggregations.
+     */
+    public int getStreamingAggregationMinShardSize() {
+        return streamingAggregationMinShardSize;
+    }
+
+    private void setStreamingAggregationMinShardSize(int streamingAggregationMinShardSize) {
+        this.streamingAggregationMinShardSize = streamingAggregationMinShardSize;
+    }
+
     /**
      * Returns the index sort config that should be used for this index.
      */