Increase default maxMergeAtOnce from 10 to 30

Author: linuxpi

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

@@ -297,6 +297,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                 IndicesQueryCache.INDICES_CACHE_QUERY_SIZE_SETTING,
                 IndicesQueryCache.INDICES_CACHE_QUERY_COUNT_SETTING,
                 IndicesQueryCache.INDICES_QUERIES_CACHE_ALL_SEGMENTS_SETTING,
+                IndicesService.CLUSTER_DEFAULT_INDEX_MAX_MERGE_AT_ONCE_SETTING,
                 IndicesService.CLUSTER_DEFAULT_INDEX_REFRESH_INTERVAL_SETTING,
                 IndicesService.CLUSTER_MINIMUM_INDEX_REFRESH_INTERVAL_SETTING,
                 IndicesService.INDICES_ID_FIELD_DATA_ENABLED_SETTING,

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

@@ -631,7 +631,8 @@ public IndexService newIndexService(
         BiFunction<IndexSettings, ShardRouting, TranslogFactory> translogFactorySupplier,
         Supplier<TimeValue> clusterDefaultRefreshIntervalSupplier,
         RecoverySettings recoverySettings,
-        RemoteStoreSettings remoteStoreSettings
+        RemoteStoreSettings remoteStoreSettings,
+        Supplier<Integer> clusterDefaultMaxMergeAtOnceSupplier
     ) throws IOException {
         return newIndexService(
             indexCreationContext,
@@ -656,7 +657,8 @@ public IndexService newIndexService(
             recoverySettings,
             remoteStoreSettings,
             (s) -> {},
-            shardId -> ReplicationStats.empty()
+            shardId -> ReplicationStats.empty(),
+            clusterDefaultMaxMergeAtOnceSupplier
         );
     }
 
@@ -683,7 +685,8 @@ public IndexService newIndexService(
         RecoverySettings recoverySettings,
         RemoteStoreSettings remoteStoreSettings,
         Consumer<IndexShard> replicator,
-        Function<ShardId, ReplicationStats> segmentReplicationStatsProvider
+        Function<ShardId, ReplicationStats> segmentReplicationStatsProvider,
+        Supplier<Integer> clusterDefaultMaxMergeAtOnceSupplier
     ) throws IOException {
         final IndexEventListener eventListener = freeze();
         Function<IndexService, CheckedFunction<DirectoryReader, DirectoryReader, IOException>> readerWrapperFactory = indexReaderWrapper
@@ -746,7 +749,8 @@ public IndexService newIndexService(
                 fileCache,
                 compositeIndexSettings,
                 replicator,
-                segmentReplicationStatsProvider
+                segmentReplicationStatsProvider,
+                clusterDefaultMaxMergeAtOnceSupplier
             );
             success = true;
             return indexService;

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

@@ -198,6 +198,7 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private final CompositeIndexSettings compositeIndexSettings;
     private final Consumer<IndexShard> replicator;
     private final Function<ShardId, ReplicationStats> segmentReplicationStatsProvider;
+    private final Supplier<Integer> clusterDefaultMaxMergeAtOnce;
 
     public IndexService(
         IndexSettings indexSettings,
@@ -237,7 +238,8 @@ public IndexService(
         FileCache fileCache,
         CompositeIndexSettings compositeIndexSettings,
         Consumer<IndexShard> replicator,
-        Function<ShardId, ReplicationStats> segmentReplicationStatsProvider
+        Function<ShardId, ReplicationStats> segmentReplicationStatsProvider,
+        Supplier<Integer> clusterDefaultMaxMergeAtOnceSupplier
     ) {
         super(indexSettings);
         this.allowExpensiveQueries = allowExpensiveQueries;
@@ -325,6 +327,9 @@ public IndexService(
         this.fileCache = fileCache;
         this.replicator = replicator;
         this.segmentReplicationStatsProvider = segmentReplicationStatsProvider;
+        this.clusterDefaultMaxMergeAtOnce = clusterDefaultMaxMergeAtOnceSupplier;
+
+        indexSettings.setMaxMergesAtOnce(this.clusterDefaultMaxMergeAtOnce.get());
         updateFsyncTaskIfNecessary();
     }
 
@@ -362,7 +367,8 @@ public IndexService(
         BiFunction<IndexSettings, ShardRouting, TranslogFactory> translogFactorySupplier,
         Supplier<TimeValue> clusterDefaultRefreshIntervalSupplier,
         RecoverySettings recoverySettings,
-        RemoteStoreSettings remoteStoreSettings
+        RemoteStoreSettings remoteStoreSettings,
+        Supplier<Integer> clusterDefaultMaxMergeAtOnce
     ) {
         this(
             indexSettings,
@@ -402,7 +408,8 @@ public IndexService(
             null,
             null,
             s -> {},
-            (shardId) -> ReplicationStats.empty()
+            (shardId) -> ReplicationStats.empty(),
+            clusterDefaultMaxMergeAtOnce
         );
     }
 
@@ -1122,11 +1129,12 @@ private void updateReplicationTask() {
     }
 
     /**
-     * Called whenever the refresh interval changes. This can happen in 2 cases -
-     * 1. {@code cluster.default.index.refresh_interval} cluster setting changes. The change would only happen for
-     * indexes relying on cluster default.
-     * 2. {@code index.refresh_interval} index setting changes.
+     * Called whenever the cluster level {@code cluster.default.index.max_merge_at_once} changes.
      */
+    public void onDefaultMaxMergeAtOnceChanged(int defaultMaxMergeAtOnce) {
+        indexSettings.setMaxMergesAtOnce(defaultMaxMergeAtOnce);
+    }
+
     public void onRefreshIntervalChange() {
         if (refreshTask.getInterval().equals(getRefreshInterval())) {
             return;

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

@@ -1250,6 +1250,15 @@ private void setRefreshInterval(TimeValue timeValue) {
         this.refreshInterval = timeValue;
     }
 
+    /**
+    * Update the maxMergesAtOnce if no index level override exists in index level settings
+    */
+    void setMaxMergesAtOnce(int newMaxMergesAtOnce) {
+        if (TieredMergePolicyProvider.INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_SETTING.exists(getSettings()) == false) {
+            tieredMergePolicyProvider.setMaxMergesAtOnce(newMaxMergesAtOnce);
+        }
+    }
+
     /**
      * Returns the settings for this index. These settings contain the node and index level settings where
      * settings that are specified on both index and node level are overwritten by the index settings.

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

@@ -143,7 +143,9 @@ public final class TieredMergePolicyProvider implements MergePolicyProvider {
      */
     public static final ByteSizeValue DEFAULT_FLOOR_SEGMENT = new ByteSizeValue(16, ByteSizeUnit.MB);
 
-    public static final int DEFAULT_MAX_MERGE_AT_ONCE = 10;
+    public static final int MIN_DEFAULT_MAX_MERGE_AT_ONCE = 2;
+    public static final int DEFAULT_MAX_MERGE_AT_ONCE = 30;
+
     public static final ByteSizeValue DEFAULT_MAX_MERGED_SEGMENT = new ByteSizeValue(5, ByteSizeUnit.GB);
     public static final double DEFAULT_SEGMENTS_PER_TIER = 10.0d;
     public static final double DEFAULT_RECLAIM_DELETES_WEIGHT = 2.0d;
@@ -173,7 +175,7 @@ public final class TieredMergePolicyProvider implements MergePolicyProvider {
     public static final Setting<Integer> INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_SETTING = Setting.intSetting(
         "index.merge.policy.max_merge_at_once",
         DEFAULT_MAX_MERGE_AT_ONCE,
-        2,
+        MIN_DEFAULT_MAX_MERGE_AT_ONCE,
         Property.Dynamic,
         Property.IndexScope
     );
@@ -225,7 +227,7 @@ public final class TieredMergePolicyProvider implements MergePolicyProvider {
                 INDEX_MERGE_ENABLED
             );
         }
-        maxMergeAtOnce = adjustMaxMergeAtOnceIfNeeded(maxMergeAtOnce, segmentsPerTier);
+//        maxMergeAtOnce = adjustMaxMergeAtOnceIfNeeded(maxMergeAtOnce, segmentsPerTier);
         tieredMergePolicy.setNoCFSRatio(indexSettings.getValue(INDEX_COMPOUND_FORMAT_SETTING));
         tieredMergePolicy.setForceMergeDeletesPctAllowed(forceMergeDeletesPctAllowed);
         tieredMergePolicy.setFloorSegmentMB(floorSegment.getMbFrac());

server/src/main/java/org/opensearch/indices/IndicesService.java

@@ -209,6 +209,8 @@
 import static org.opensearch.core.common.util.CollectionUtils.arrayAsArrayList;
 import static org.opensearch.index.IndexService.IndexCreationContext.CREATE_INDEX;
 import static org.opensearch.index.IndexService.IndexCreationContext.METADATA_VERIFICATION;
+import static org.opensearch.index.TieredMergePolicyProvider.DEFAULT_MAX_MERGE_AT_ONCE;
+import static org.opensearch.index.TieredMergePolicyProvider.MIN_DEFAULT_MAX_MERGE_AT_ONCE;
 import static org.opensearch.index.query.AbstractQueryBuilder.parseInnerQueryBuilder;
 import static org.opensearch.indices.IndicesRequestCache.INDICES_REQUEST_CACHE_MAX_SIZE_ALLOWED_IN_CACHE_SETTING;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteDataAttributePresent;
@@ -275,6 +277,21 @@ public class IndicesService extends AbstractLifecycleComponent
         Property.Dynamic
     );
 
+    /**
+     * This setting is used to set the maxMergeAtOnce parameter for {@code TieredMergePolicy}
+     * when the {@code index.merge.policy.max_merge_at_once} index setting is not provided during index creation
+     * or when the existing {@code index.merge.policy.max_merge_at_once} index setting is set as null.
+     * This comes handy when the user wants to change the maxMergeAtOnce across all indexes created in a cluster
+     * which is different from the default.
+     */
+    public static final Setting<Integer> CLUSTER_DEFAULT_INDEX_MAX_MERGE_AT_ONCE_SETTING = Setting.intSetting(
+        "cluster.default.index.max_merge_at_once",
+        DEFAULT_MAX_MERGE_AT_ONCE,
+        MIN_DEFAULT_MAX_MERGE_AT_ONCE,
+        Property.NodeScope,
+        Property.Dynamic
+    );
+
     /**
      * This setting is used to set the minimum refresh interval applicable for all indexes in a cluster. The
      * {@code cluster.default.index.refresh_interval} setting value needs to be higher than this setting's value. Index
@@ -369,6 +386,7 @@ public class IndicesService extends AbstractLifecycleComponent
     private final Consumer<IndexShard> replicator;
     private final Function<ShardId, ReplicationStats> segmentReplicationStatsProvider;
     private volatile int maxSizeInRequestCache;
+    private volatile int defaultMaxMergeAtOnce;
 
     @Override
     protected void doStart() {
@@ -523,6 +541,10 @@ protected void closeInternal() {
         this.maxSizeInRequestCache = INDICES_REQUEST_CACHE_MAX_SIZE_ALLOWED_IN_CACHE_SETTING.get(clusterService.getSettings());
         clusterService.getClusterSettings()
             .addSettingsUpdateConsumer(INDICES_REQUEST_CACHE_MAX_SIZE_ALLOWED_IN_CACHE_SETTING, this::setMaxSizeInRequestCache);
+
+        this.defaultMaxMergeAtOnce = CLUSTER_DEFAULT_INDEX_MAX_MERGE_AT_ONCE_SETTING.get(clusterService.getSettings());
+        clusterService.getClusterSettings()
+            .addSettingsUpdateConsumer(CLUSTER_DEFAULT_INDEX_MAX_MERGE_AT_ONCE_SETTING, this::onDefaultMaxMergeAtOnceUpdate);
     }
 
     public IndicesService(
@@ -607,6 +629,22 @@ private void onRefreshIntervalUpdate(TimeValue clusterDefaultRefreshInterval) {
         }
     }
 
+    /**
+     * The changes to dynamic cluster setting {@code cluster.default.index.max_merge_at_once} needs to be updated. This
+     * method gets called whenever the setting changes. We set the instance variable with the updated value as this is
+     * also a supplier to all IndexService that have been created on the node. We also notify the change to all
+     * IndexService instances that are created on this node.
+     *
+     * @param newDefaultMaxMergeAtOnce the updated cluster default maxMergeAtOnce.
+     */
+    private void onDefaultMaxMergeAtOnceUpdate(int newDefaultMaxMergeAtOnce) {
+        this.defaultMaxMergeAtOnce = newDefaultMaxMergeAtOnce; // do we need this?
+        for (Map.Entry<String, IndexService> entry : indices.entrySet()) {
+            IndexService indexService = entry.getValue();
+            indexService.onDefaultMaxMergeAtOnceChanged(newDefaultMaxMergeAtOnce);
+        }
+    }
+
     private static BiFunction<IndexSettings, ShardRouting, TranslogFactory> getTranslogFactorySupplier(
         Supplier<RepositoriesService> repositoriesServiceSupplier,
         ThreadPool threadPool,
@@ -1009,7 +1047,8 @@ private synchronized IndexService createIndexService(
             this.recoverySettings,
             this.remoteStoreSettings,
             replicator,
-            segmentReplicationStatsProvider
+            segmentReplicationStatsProvider,
+            this::getClusterDefaultMaxMergeAtOnce
         );
     }
 
@@ -2155,6 +2194,10 @@ private TimeValue getClusterDefaultRefreshInterval() {
         return this.clusterDefaultRefreshInterval;
     }
 
+    private Integer getClusterDefaultMaxMergeAtOnce() {
+        return this.defaultMaxMergeAtOnce;
+    }
+
     public RemoteStoreSettings getRemoteStoreSettings() {
         return this.remoteStoreSettings;
     }

server/src/test/java/org/opensearch/index/IndexModuleTests.java

@@ -266,7 +266,8 @@ private IndexService newIndexService(IndexModule module) throws IOException {
             DefaultRecoverySettings.INSTANCE,
             DefaultRemoteStoreSettings.INSTANCE,
             s -> {},
-            null
+            null,
+            () -> TieredMergePolicyProvider.DEFAULT_MAX_MERGE_AT_ONCE
         );
     }