Add an index setting to control TieredMergePolicy#deletesPctAllowed (#32907)

This change adds an expert index setting called `index.merge.policy.deletes_pct_allowed`.
It controls the maximum percentage of deleted documents that is tolerated in the index.
Lower values make the index more space efficient at the expense of increased CPU and I/O activity.
Values must be between `20` and `50`. Default value is `33`.

Author: jimczi

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

@@ -96,6 +96,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         IndexingSlowLog.INDEX_INDEXING_SLOWLOG_REFORMAT_SETTING,
         IndexingSlowLog.INDEX_INDEXING_SLOWLOG_MAX_SOURCE_CHARS_TO_LOG_SETTING,
         MergePolicyConfig.INDEX_COMPOUND_FORMAT_SETTING,
+        MergePolicyConfig.INDEX_MERGE_POLICY_DELETES_PCT_ALLOWED_SETTING,
         MergePolicyConfig.INDEX_MERGE_POLICY_EXPUNGE_DELETES_ALLOWED_SETTING,
         MergePolicyConfig.INDEX_MERGE_POLICY_FLOOR_SEGMENT_SETTING,
         MergePolicyConfig.INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_SETTING,

server/src/main/java/org/elasticsearch/common/settings/Setting.java

@@ -1303,11 +1303,18 @@ public static Setting<TimeValue> positiveTimeSetting(String key, TimeValue defau
     }
 
     public static Setting<Double> doubleSetting(String key, double defaultValue, double minValue, Property... properties) {
+        return doubleSetting(key, defaultValue, minValue, Double.POSITIVE_INFINITY, properties);
+    }
+
+    public static Setting<Double> doubleSetting(String key, double defaultValue, double minValue, double maxValue, Property... properties) {
         return new Setting<>(key, (s) -> Double.toString(defaultValue), (s) -> {
             final double d = Double.parseDouble(s);
             if (d < minValue) {
                 throw new IllegalArgumentException("Failed to parse value [" + s + "] for setting [" + key + "] must be >= " + minValue);
             }
+            if (d > maxValue) {
+                throw new IllegalArgumentException("Failed to parse value [" + s + "] for setting [" + key + "] must be <= " + maxValue);
+            }
             return d;
         }, properties);
     }

server/src/main/java/org/elasticsearch/index/EsTieredMergePolicy.java

@@ -108,4 +108,13 @@ public void setSegmentsPerTier(double segmentsPerTier) {
     public double getSegmentsPerTier() {
         return regularMergePolicy.getSegmentsPerTier();
     }
+
+    public void setDeletesPctAllowed(double deletesPctAllowed) {
+        regularMergePolicy.setDeletesPctAllowed(deletesPctAllowed);
+        forcedMergePolicy.setDeletesPctAllowed(deletesPctAllowed);
+    }
+
+    public double getDeletesPctAllowed() {
+        return regularMergePolicy.getDeletesPctAllowed();
+    }
 }

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

@@ -466,6 +466,7 @@ public IndexSettings(final IndexMetaData indexMetaData, final Settings nodeSetti
         defaultPipeline = scopedSettings.get(DEFAULT_PIPELINE);
 
         scopedSettings.addSettingsUpdateConsumer(MergePolicyConfig.INDEX_COMPOUND_FORMAT_SETTING, mergePolicyConfig::setNoCFSRatio);
+        scopedSettings.addSettingsUpdateConsumer(MergePolicyConfig.INDEX_MERGE_POLICY_DELETES_PCT_ALLOWED_SETTING, mergePolicyConfig::setDeletesPctAllowed);
         scopedSettings.addSettingsUpdateConsumer(MergePolicyConfig.INDEX_MERGE_POLICY_EXPUNGE_DELETES_ALLOWED_SETTING, mergePolicyConfig::setExpungeDeletesAllowed);
         scopedSettings.addSettingsUpdateConsumer(MergePolicyConfig.INDEX_MERGE_POLICY_FLOOR_SEGMENT_SETTING, mergePolicyConfig::setFloorSegmentSetting);
         scopedSettings.addSettingsUpdateConsumer(MergePolicyConfig.INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_SETTING, mergePolicyConfig::setMaxMergesAtOnce);

server/src/main/java/org/elasticsearch/index/MergePolicyConfig.java

@@ -83,11 +83,12 @@
  *     &gt;= than the <code>max_merge_at_once</code> otherwise you'll force too many merges to
  *     occur.
  *
- * <li><code>index.merge.policy.reclaim_deletes_weight</code>:
+ * <li><code>index.merge.policy.deletes_pct_allowed</code>:
  *
- *     Controls how aggressively merges that reclaim more deletions are favored.
- *     Higher values favor selecting merges that reclaim deletions. A value of
- *     <code>0.0</code> means deletions don't impact merge selection. Defaults to <code>2.0</code>.
+ *     Controls the maximum percentage of deleted documents that is tolerated in
+ *     the index. Lower values make the index more space efficient at the
+ *     expense of increased CPU and I/O activity. Values must be between <code>20</code> and
+ *     <code>50</code>. Default value is <code>33</code>.
  * </ul>
  *
  * <p>
@@ -127,6 +128,7 @@ public final class MergePolicyConfig {
     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;
+    public static final double          DEFAULT_DELETES_PCT_ALLOWED         = 33.0d;
     public static final Setting<Double> INDEX_COMPOUND_FORMAT_SETTING       =
         new Setting<>("index.compound_format", Double.toString(TieredMergePolicy.DEFAULT_NO_CFS_RATIO), MergePolicyConfig::parseNoCFSRatio,
             Property.Dynamic, Property.IndexScope);
@@ -152,6 +154,9 @@ public final class MergePolicyConfig {
     public static final Setting<Double> INDEX_MERGE_POLICY_RECLAIM_DELETES_WEIGHT_SETTING =
         Setting.doubleSetting("index.merge.policy.reclaim_deletes_weight", DEFAULT_RECLAIM_DELETES_WEIGHT, 0.0d,
             Property.Dynamic, Property.IndexScope, Property.Deprecated);
+    public static final Setting<Double> INDEX_MERGE_POLICY_DELETES_PCT_ALLOWED_SETTING =
+        Setting.doubleSetting("index.merge.policy.deletes_pct_allowed", DEFAULT_DELETES_PCT_ALLOWED, 20.0d, 50.0d,
+            Property.Dynamic, Property.IndexScope);
     public static final String INDEX_MERGE_ENABLED = "index.merge.enabled"; // don't convert to Setting<> and register... we only set this in tests and register via a plugin
 
 
@@ -165,6 +170,7 @@ public final class MergePolicyConfig {
         ByteSizeValue maxMergedSegment = indexSettings.getValue(INDEX_MERGE_POLICY_MAX_MERGED_SEGMENT_SETTING);
         double segmentsPerTier = indexSettings.getValue(INDEX_MERGE_POLICY_SEGMENTS_PER_TIER_SETTING);
         double reclaimDeletesWeight = indexSettings.getValue(INDEX_MERGE_POLICY_RECLAIM_DELETES_WEIGHT_SETTING);
+        double deletesPctAllowed = indexSettings.getValue(INDEX_MERGE_POLICY_DELETES_PCT_ALLOWED_SETTING);
         this.mergesEnabled = indexSettings.getSettings()
             .getAsBooleanLenientForPreEs6Indices(indexSettings.getIndexVersionCreated(), INDEX_MERGE_ENABLED, true, new DeprecationLogger(logger));
         if (mergesEnabled == false) {
@@ -178,9 +184,10 @@ public final class MergePolicyConfig {
         mergePolicy.setMaxMergeAtOnceExplicit(maxMergeAtOnceExplicit);
         mergePolicy.setMaxMergedSegmentMB(maxMergedSegment.getMbFrac());
         mergePolicy.setSegmentsPerTier(segmentsPerTier);
+        mergePolicy.setDeletesPctAllowed(deletesPctAllowed);
         if (logger.isTraceEnabled()) {
-            logger.trace("using [tiered] merge mergePolicy with expunge_deletes_allowed[{}], floor_segment[{}], max_merge_at_once[{}], max_merge_at_once_explicit[{}], max_merged_segment[{}], segments_per_tier[{}], reclaim_deletes_weight[{}]",
-                forceMergeDeletesPctAllowed, floorSegment, maxMergeAtOnce, maxMergeAtOnceExplicit, maxMergedSegment, segmentsPerTier, reclaimDeletesWeight);
+            logger.trace("using [tiered] merge mergePolicy with expunge_deletes_allowed[{}], floor_segment[{}], max_merge_at_once[{}], max_merge_at_once_explicit[{}], max_merged_segment[{}], segments_per_tier[{}], deletes_pct_allowed[{}]",
+                forceMergeDeletesPctAllowed, floorSegment, maxMergeAtOnce, maxMergeAtOnceExplicit, maxMergedSegment, segmentsPerTier, deletesPctAllowed);
         }
     }
 
@@ -212,6 +219,10 @@ void setNoCFSRatio(Double noCFSRatio) {
         mergePolicy.setNoCFSRatio(noCFSRatio);
     }
 
+    void setDeletesPctAllowed(Double deletesPctAllowed) {
+        mergePolicy.setDeletesPctAllowed(deletesPctAllowed);
+    }
+
     private int adjustMaxMergeAtOnceIfNeeded(int maxMergeAtOnce, double segmentsPerTier) {
         // fixing maxMergeAtOnce, see TieredMergePolicy#setMaxMergeAtOnce
         if (!(segmentsPerTier >= maxMergeAtOnce)) {

server/src/test/java/org/elasticsearch/index/EsTieredMergePolicyTests.java

@@ -69,4 +69,10 @@ public void testSetSegmentsPerTier() {
         policy.setSegmentsPerTier(42);
         assertEquals(42, policy.regularMergePolicy.getSegmentsPerTier(), 0);
     }
+
+    public void testSetDeletesPctAllowed() {
+        EsTieredMergePolicy policy = new EsTieredMergePolicy();
+        policy.setDeletesPctAllowed(42);
+        assertEquals(42, policy.regularMergePolicy.getDeletesPctAllowed(), 0);
+    }
 }

server/src/test/java/org/elasticsearch/index/MergePolicySettingsTests.java

@@ -29,6 +29,7 @@
 
 import static org.elasticsearch.common.settings.Settings.Builder.EMPTY_SETTINGS;
 import static org.elasticsearch.index.IndexSettingsTests.newIndexMeta;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
 public class MergePolicySettingsTests extends ESTestCase {
@@ -100,13 +101,22 @@ public void testTieredMergePolicySettingsUpdate() throws IOException {
         indexSettings.updateIndexMetaData(newIndexMeta("index", Settings.builder().put(MergePolicyConfig.INDEX_MERGE_POLICY_SEGMENTS_PER_TIER_SETTING.getKey(), MergePolicyConfig.DEFAULT_SEGMENTS_PER_TIER + 1).build()));
         assertEquals(((EsTieredMergePolicy) indexSettings.getMergePolicy()).getSegmentsPerTier(), MergePolicyConfig.DEFAULT_SEGMENTS_PER_TIER + 1, 0);
 
+        assertEquals(((EsTieredMergePolicy) indexSettings.getMergePolicy()).getDeletesPctAllowed(), MergePolicyConfig.DEFAULT_DELETES_PCT_ALLOWED, 0);
+        indexSettings.updateIndexMetaData(newIndexMeta("index", Settings.builder().put(MergePolicyConfig.INDEX_MERGE_POLICY_DELETES_PCT_ALLOWED_SETTING.getKey(), 22).build()));
+        assertEquals(((EsTieredMergePolicy) indexSettings.getMergePolicy()).getDeletesPctAllowed(), 22, 0);
+
+        IllegalArgumentException exc = expectThrows(IllegalArgumentException.class, () ->
+            indexSettings.updateIndexMetaData(newIndexMeta("index", Settings.builder().put(MergePolicyConfig.INDEX_MERGE_POLICY_DELETES_PCT_ALLOWED_SETTING.getKey(), 53).build())));
+        final Throwable cause = exc.getCause();
+        assertThat(cause.getMessage(), containsString("must be <= 50.0"));
         indexSettings.updateIndexMetaData(newIndexMeta("index", EMPTY_SETTINGS)); // see if defaults are restored
         assertEquals(((EsTieredMergePolicy) indexSettings.getMergePolicy()).getForceMergeDeletesPctAllowed(), MergePolicyConfig.DEFAULT_EXPUNGE_DELETES_ALLOWED, 0.0d);
         assertEquals(((EsTieredMergePolicy) indexSettings.getMergePolicy()).getFloorSegmentMB(), new ByteSizeValue(MergePolicyConfig.DEFAULT_FLOOR_SEGMENT.getMb(), ByteSizeUnit.MB).getMbFrac(), 0.00);
         assertEquals(((EsTieredMergePolicy) indexSettings.getMergePolicy()).getMaxMergeAtOnce(), MergePolicyConfig.DEFAULT_MAX_MERGE_AT_ONCE);
         assertEquals(((EsTieredMergePolicy) indexSettings.getMergePolicy()).getMaxMergeAtOnceExplicit(), MergePolicyConfig.DEFAULT_MAX_MERGE_AT_ONCE_EXPLICIT);
         assertEquals(((EsTieredMergePolicy) indexSettings.getMergePolicy()).getMaxMergedSegmentMB(), new ByteSizeValue(MergePolicyConfig.DEFAULT_MAX_MERGED_SEGMENT.getBytes() + 1).getMbFrac(), 0.0001);
         assertEquals(((EsTieredMergePolicy) indexSettings.getMergePolicy()).getSegmentsPerTier(), MergePolicyConfig.DEFAULT_SEGMENTS_PER_TIER, 0);
+        assertEquals(((EsTieredMergePolicy) indexSettings.getMergePolicy()).getDeletesPctAllowed(), MergePolicyConfig.DEFAULT_DELETES_PCT_ALLOWED, 0);
     }
 
     public Settings build(String value) {