Use LogByteSizeMergePolicy instead of TieredMergePolicy for time-based data.

docs/changelog/92684.yaml

@@ -0,0 +1,6 @@
+pr: 92684
+summary: Use `LogByteSizeMergePolicy` instead of `TieredMergePolicy` for time-based
+  data
+area: Engine
+type: enhancement
+issues: []

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

@@ -89,13 +89,15 @@ 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_TYPE_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,
         MergePolicyConfig.INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_EXPLICIT_SETTING,
         MergePolicyConfig.INDEX_MERGE_POLICY_MAX_MERGED_SEGMENT_SETTING,
         MergePolicyConfig.INDEX_MERGE_POLICY_SEGMENTS_PER_TIER_SETTING,
+        MergePolicyConfig.INDEX_MERGE_POLICY_MERGE_FACTOR_SETTING,
         IndexSortConfig.INDEX_SORT_FIELD_SETTING,
         IndexSortConfig.INDEX_SORT_ORDER_SETTING,
         IndexSortConfig.INDEX_SORT_MISSING_SETTING,

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

@@ -784,6 +784,7 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
             MergePolicyConfig.INDEX_COMPOUND_FORMAT_SETTING,
             mergePolicyConfig::setCompoundFormatThreshold
         );
+        scopedSettings.addSettingsUpdateConsumer(MergePolicyConfig.INDEX_MERGE_POLICY_TYPE_SETTING, mergePolicyConfig::setMergePolicyType);
         scopedSettings.addSettingsUpdateConsumer(
             MergePolicyConfig.INDEX_MERGE_POLICY_DELETES_PCT_ALLOWED_SETTING,
             mergePolicyConfig::setDeletesPctAllowed
@@ -809,6 +810,10 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
             MergePolicyConfig.INDEX_MERGE_POLICY_SEGMENTS_PER_TIER_SETTING,
             mergePolicyConfig::setSegmentsPerTier
         );
+        scopedSettings.addSettingsUpdateConsumer(
+            MergePolicyConfig.INDEX_MERGE_POLICY_MERGE_FACTOR_SETTING,
+            mergePolicyConfig::setMergeFactor
+        );
 
         scopedSettings.addSettingsUpdateConsumer(
             MergeSchedulerConfig.MAX_THREAD_COUNT_SETTING,
@@ -1216,8 +1221,8 @@ public long getGcDeletesInMillis() {
     /**
      * Returns the merge policy that should be used for this index.
      */
-    public MergePolicy getMergePolicy() {
-        return mergePolicyConfig.getMergePolicy();
+    public MergePolicy getMergePolicy(boolean isTimeBasedIndex) {
+        return mergePolicyConfig.getMergePolicy(isTimeBasedIndex);
     }
 
     public <T> T getValue(Setting<T> setting) {

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

@@ -9,6 +9,7 @@
 package org.elasticsearch.index;
 
 import org.apache.logging.log4j.Logger;
+import org.apache.lucene.index.LogByteSizeMergePolicy;
 import org.apache.lucene.index.MergePolicy;
 import org.apache.lucene.index.NoMergePolicy;
 import org.apache.lucene.index.TieredMergePolicy;
@@ -101,15 +102,31 @@
  */
 
 public final class MergePolicyConfig {
-    private final TieredMergePolicy mergePolicy = new TieredMergePolicy();
+    private final TieredMergePolicy tieredMergePolicy = new TieredMergePolicy();
+    private final LogByteSizeMergePolicy logByteSizeMergePolicy = new LogByteSizeMergePolicy();
     private final Logger logger;
     private final boolean mergesEnabled;
+    private volatile Type mergePolicyType;
 
     public static final double DEFAULT_EXPUNGE_DELETES_ALLOWED = 10d;
     public static final ByteSizeValue DEFAULT_FLOOR_SEGMENT = new ByteSizeValue(2, ByteSizeUnit.MB);
     public static final int DEFAULT_MAX_MERGE_AT_ONCE = 10;
     public static final ByteSizeValue DEFAULT_MAX_MERGED_SEGMENT = new ByteSizeValue(5, ByteSizeUnit.GB);
+    /**
+     * Time-based data generally gets rolled over, so there is not much value in enforcing a maximum segment size, which has the side effect
+     * of merging fewer segments together than the merge factor, which in-turn increases write amplification. So we set an arbitrarily high
+     * roof that serves as a protection that we expect to never hit.
+     */
+    public static final ByteSizeValue DEFAULT_MAX_TIME_BASED_MERGED_SEGMENT = new ByteSizeValue(100, ByteSizeUnit.GB);
     public static final double DEFAULT_SEGMENTS_PER_TIER = 10.0d;
+    /**
+     * A default value for {@link LogByteSizeMergePolicy}'s merge factor: 16. This default value differs from the Lucene default of 10 in
+     * order to account for the fact that Elasticsearch uses {@link LogByteSizeMergePolicy} for time-based data, where it usually makes
+     * sense to merge data less aggressively, and because {@link LogByteSizeMergePolicy} merges segments more aggressively than
+     * {@link TieredMergePolicy} for the same number of segments per tier / merge factor because {@link TieredMergePolicy} makes decisions
+     * at the whole index level, while {@link LogByteSizeMergePolicy} makes decisions on a per-tier basis.
+     */
+    public static final int DEFAULT_MERGE_FACTOR = 16;
     public static final double DEFAULT_DELETES_PCT_ALLOWED = 20.0d;
     private static final String INDEX_COMPOUND_FORMAT_SETTING_KEY = "index.compound_format";
     public static final Setting<CompoundFileThreshold> INDEX_COMPOUND_FORMAT_SETTING = new Setting<>(
@@ -120,6 +137,57 @@ public final class MergePolicyConfig {
         Property.IndexScope
     );
 
+    public enum Type {
+        UNSET {
+            @Override
+            MergePolicy getMergePolicy(MergePolicyConfig config, boolean isTimeBasedIndex) {
+                if (isTimeBasedIndex) {
+                    // With time-based data, it's important that the merge policy only merges adjacent segments, so that segments end up
+                    // with non-overlapping time ranges if data gets indexed in order. This makes queries more efficient, as range filters
+                    // on the timestamp are more likely to either fully match a segment or not match it at all, which Lucene handles more
+                    // efficiently than a partially matching segment. This also plays nicely with the fact that recent data is more heavily
+                    // queried than older data, so some segments are more likely to not get touched at all by queries if they don't
+                    // intersect with the query's range.
+
+                    // The downside of only doing adjacent merges is that it may result in slightly less efficient merging if there is a lot
+                    // of variance in the size of flushes. Allowing merges of non-adjacent segments also makes it possible to reclaim
+                    // deletes a bit more efficiently by merging together segments that have the most deletes, even though they might not be
+                    // adjacent. But overall, the benefits of only doing adjacent merging exceed the downsides for time-based data.
+
+                    // LogByteSizeMergePolicy is similar to TieredMergePolicy, as it also tries to organize segments into tiers of
+                    // exponential sizes. The main difference is that it never merges non-adjacent segments, which is an interesting
+                    // property for time-based data as described above.
+
+                    return config.logByteSizeMergePolicy;
+                } else {
+                    return config.tieredMergePolicy;
+                }
+            }
+        },
+        TIERED {
+            @Override
+            MergePolicy getMergePolicy(MergePolicyConfig config, boolean isTimeBasedIndex) {
+                return config.tieredMergePolicy;
+            }
+        },
+        LOG_BYTE_SIZE {
+            @Override
+            MergePolicy getMergePolicy(MergePolicyConfig config, boolean isTimeBasedIndex) {
+                return config.logByteSizeMergePolicy;
+            }
+        };
+
+        abstract MergePolicy getMergePolicy(MergePolicyConfig config, boolean isTimeSeries);
+    }
+
+    public static final Setting<Type> INDEX_MERGE_POLICY_TYPE_SETTING = Setting.enumSetting(
+        Type.class,
+        "index.merge.policy.type",
+        Type.UNSET,
+        Property.Dynamic,
+        Property.IndexScope
+    );
+
     public static final Setting<Double> INDEX_MERGE_POLICY_EXPUNGE_DELETES_ALLOWED_SETTING = Setting.doubleSetting(
         "index.merge.policy.expunge_deletes_allowed",
         DEFAULT_EXPUNGE_DELETES_ALLOWED,
@@ -150,7 +218,8 @@ public final class MergePolicyConfig {
     );
     public static final Setting<ByteSizeValue> INDEX_MERGE_POLICY_MAX_MERGED_SEGMENT_SETTING = Setting.byteSizeSetting(
         "index.merge.policy.max_merged_segment",
-        DEFAULT_MAX_MERGED_SEGMENT,
+        // We're not using DEFAULT_MAX_MERGED_SEGMENT here as we want different defaults for time-based data vs. non-time based
+        new ByteSizeValue(0, ByteSizeUnit.BYTES),
         Property.Dynamic,
         Property.IndexScope
     );
@@ -161,6 +230,13 @@ public final class MergePolicyConfig {
         Property.Dynamic,
         Property.IndexScope
     );
+    public static final Setting<Integer> INDEX_MERGE_POLICY_MERGE_FACTOR_SETTING = Setting.intSetting(
+        "index.merge.policy.merge_factor",
+        DEFAULT_MERGE_FACTOR,
+        2,
+        Property.Dynamic,
+        Property.IndexScope
+    );
     public static final Setting<Double> INDEX_MERGE_POLICY_DELETES_PCT_ALLOWED_SETTING = Setting.doubleSetting(
         "index.merge.policy.deletes_pct_allowed",
         DEFAULT_DELETES_PCT_ALLOWED,
@@ -174,13 +250,15 @@ public final class MergePolicyConfig {
 
     MergePolicyConfig(Logger logger, IndexSettings indexSettings) {
         this.logger = logger;
+        Type mergePolicyType = indexSettings.getValue(INDEX_MERGE_POLICY_TYPE_SETTING);
         double forceMergeDeletesPctAllowed = indexSettings.getValue(INDEX_MERGE_POLICY_EXPUNGE_DELETES_ALLOWED_SETTING); // percentage
         ByteSizeValue floorSegment = indexSettings.getValue(INDEX_MERGE_POLICY_FLOOR_SEGMENT_SETTING);
         int maxMergeAtOnce = indexSettings.getValue(INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_SETTING);
         // TODO is this really a good default number for max_merge_segment, what happens for large indices,
         // won't they end up with many segments?
         ByteSizeValue maxMergedSegment = indexSettings.getValue(INDEX_MERGE_POLICY_MAX_MERGED_SEGMENT_SETTING);
         double segmentsPerTier = indexSettings.getValue(INDEX_MERGE_POLICY_SEGMENTS_PER_TIER_SETTING);
+        int mergeFactor = indexSettings.getValue(INDEX_MERGE_POLICY_MERGE_FACTOR_SETTING);
         double deletesPctAllowed = indexSettings.getValue(INDEX_MERGE_POLICY_DELETES_PCT_ALLOWED_SETTING);
         this.mergesEnabled = indexSettings.getSettings().getAsBoolean(INDEX_MERGE_ENABLED, true);
         if (mergesEnabled == false) {
@@ -190,15 +268,17 @@ public final class MergePolicyConfig {
             );
         }
         maxMergeAtOnce = adjustMaxMergeAtOnceIfNeeded(maxMergeAtOnce, segmentsPerTier);
-        indexSettings.getValue(INDEX_COMPOUND_FORMAT_SETTING).configure(mergePolicy);
-        mergePolicy.setForceMergeDeletesPctAllowed(forceMergeDeletesPctAllowed);
-        mergePolicy.setFloorSegmentMB(floorSegment.getMbFrac());
-        mergePolicy.setMaxMergeAtOnce(maxMergeAtOnce);
-        mergePolicy.setMaxMergedSegmentMB(maxMergedSegment.getMbFrac());
-        mergePolicy.setSegmentsPerTier(segmentsPerTier);
-        mergePolicy.setDeletesPctAllowed(deletesPctAllowed);
+        setMergePolicyType(mergePolicyType);
+        setCompoundFormatThreshold(indexSettings.getValue(INDEX_COMPOUND_FORMAT_SETTING));
+        setExpungeDeletesAllowed(forceMergeDeletesPctAllowed);
+        setFloorSegmentSetting(floorSegment);
+        setMaxMergesAtOnce(maxMergeAtOnce);
+        setMaxMergedSegment(maxMergedSegment);
+        setSegmentsPerTier(segmentsPerTier);
+        setMergeFactor(mergeFactor);
+        setDeletesPctAllowed(deletesPctAllowed);
         logger.trace(
-            "using [tiered] merge mergePolicy with expunge_deletes_allowed[{}], floor_segment[{}],"
+            "using merge policy with expunge_deletes_allowed[{}], floor_segment[{}],"
                 + " max_merge_at_once[{}], max_merged_segment[{}], segments_per_tier[{}],"
                 + " deletes_pct_allowed[{}]",
             forceMergeDeletesPctAllowed,
@@ -210,32 +290,54 @@ public final class MergePolicyConfig {
         );
     }
 
-    void setSegmentsPerTier(Double segmentsPerTier) {
-        mergePolicy.setSegmentsPerTier(segmentsPerTier);
+    void setMergePolicyType(Type type) {
+        this.mergePolicyType = type;
+    }
+
+    void setSegmentsPerTier(double segmentsPerTier) {
+        tieredMergePolicy.setSegmentsPerTier(segmentsPerTier);
+        // LogByteSizeMergePolicy ignores this parameter, it always tries to have between 1 and merge_factor - 1 segments per tier.
+    }
+
+    void setMergeFactor(int mergeFactor) {
+        // TieredMergePolicy ignores this setting, it configures a number of segments per tier instead, which has different semantics.
+        logByteSizeMergePolicy.setMergeFactor(mergeFactor);
     }
 
     void setMaxMergedSegment(ByteSizeValue maxMergedSegment) {
-        mergePolicy.setMaxMergedSegmentMB(maxMergedSegment.getMbFrac());
+        // We use 0 as a placeholder for "unset".
+        if (maxMergedSegment.getBytes() == 0) {
+            tieredMergePolicy.setMaxMergedSegmentMB(DEFAULT_MAX_MERGED_SEGMENT.getMbFrac());
+            logByteSizeMergePolicy.setMaxMergeMB(DEFAULT_MAX_TIME_BASED_MERGED_SEGMENT.getMbFrac());
+        } else {
+            tieredMergePolicy.setMaxMergedSegmentMB(maxMergedSegment.getMbFrac());
+            logByteSizeMergePolicy.setMaxMergeMB(maxMergedSegment.getMbFrac());
+        }
     }
 
-    void setMaxMergesAtOnce(Integer maxMergeAtOnce) {
-        mergePolicy.setMaxMergeAtOnce(maxMergeAtOnce);
+    void setMaxMergesAtOnce(int maxMergeAtOnce) {
+        tieredMergePolicy.setMaxMergeAtOnce(maxMergeAtOnce);
+        // LogByteSizeMergePolicy ignores this parameter, it always merges merge_factor segments at once.
     }
 
     void setFloorSegmentSetting(ByteSizeValue floorSegementSetting) {
-        mergePolicy.setFloorSegmentMB(floorSegementSetting.getMbFrac());
+        tieredMergePolicy.setFloorSegmentMB(floorSegementSetting.getMbFrac());
+        logByteSizeMergePolicy.setMinMergeMB(floorSegementSetting.getMbFrac());
     }
 
     void setExpungeDeletesAllowed(Double value) {
-        mergePolicy.setForceMergeDeletesPctAllowed(value);
+        tieredMergePolicy.setForceMergeDeletesPctAllowed(value);
+        // LogByteSizeMergePolicy doesn't have a similar configuration option
     }
 
     void setCompoundFormatThreshold(CompoundFileThreshold compoundFileThreshold) {
-        compoundFileThreshold.configure(mergePolicy);
+        compoundFileThreshold.configure(tieredMergePolicy);
+        compoundFileThreshold.configure(logByteSizeMergePolicy);
     }
 
     void setDeletesPctAllowed(Double deletesPctAllowed) {
-        mergePolicy.setDeletesPctAllowed(deletesPctAllowed);
+        tieredMergePolicy.setDeletesPctAllowed(deletesPctAllowed);
+        // LogByteSizeMergePolicy doesn't have a similar configuration option
     }
 
     private int adjustMaxMergeAtOnceIfNeeded(int maxMergeAtOnce, double segmentsPerTier) {
@@ -258,8 +360,11 @@ private int adjustMaxMergeAtOnceIfNeeded(int maxMergeAtOnce, double segmentsPerT
     }
 
     @SuppressForbidden(reason = "we always use an appropriate merge scheduler alongside this policy so NoMergePolic#INSTANCE is ok")
-    MergePolicy getMergePolicy() {
-        return mergesEnabled ? mergePolicy : NoMergePolicy.INSTANCE;
+    MergePolicy getMergePolicy(boolean isTimeBasedIndex) {
+        if (mergesEnabled == false) {
+            return NoMergePolicy.INSTANCE;
+        }
+        return mergePolicyType.getMergePolicy(this, isTimeBasedIndex);
     }
 
     private static CompoundFileThreshold parseCompoundFormat(String noCFSRatio) {

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -3269,14 +3269,14 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) {
                 this.warmer.warm(reader);
             }
         };
-        final boolean isTimeseriesIndex = mapperService == null ? false : mapperService.mappingLookup().hasTimestampField();
+        final boolean isTimeBasedIndex = mapperService == null ? false : mapperService.mappingLookup().hasTimestampField();
         return new EngineConfig(
             shardId,
             threadPool,
             indexSettings,
             warmer,
             store,
-            indexSettings.getMergePolicy(),
+            indexSettings.getMergePolicy(isTimeBasedIndex),
             buildIndexAnalyzer(mapperService),
             similarityService.similarity(mapperService == null ? null : mapperService::fieldType),
             codecService,
@@ -3293,7 +3293,7 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) {
             replicationTracker::getRetentionLeases,
             this::getOperationPrimaryTerm,
             snapshotCommitSupplier,
-            isTimeseriesIndex ? TIMESERIES_LEAF_READERS_SORTER : null,
+            isTimeBasedIndex ? TIMESERIES_LEAF_READERS_SORTER : null,
             relativeTimeInNanosSupplier,
             indexCommitListener,
             routingEntry().isPromotableToPrimary()