Allow LogMergePolicy to merge more than mergeFactor segments toge…

…ther when the merge is below the min merge size.

This is essentially porting apache#266 to `LogMergePolicy`. By allowing more than
`mergeFactor` segments to be merged together for small merges, the merge policy
gets a lower write amplification and indexes have fewer small segments.

lucene/core/src/java/org/apache/lucene/index/LogMergePolicy.java

@@ -636,6 +636,28 @@ public MergeSpecification findMerges(
           mergeDocs += segmentDocs;
         }
 
+        if (end - start >= mergeFactor && mergeSize < minMergeSize && anyMerging == false) {
+          // If the merge has mergeFactor segments but is still smaller than the min merged segment
+          // size, keep packing candidate segments.
+          while (end < 1 + upto) {
+            final SegmentInfoAndLevel segLevel = levels.get(end);
+            final SegmentCommitInfo info = segLevel.info;
+            if (mergingSegments.contains(info)) {
+              anyMerging = true;
+              break;
+            }
+            long segmentSize = size(info, mergeContext);
+            long segmentDocs = sizeDocs(info, mergeContext);
+            if (mergeSize + segmentSize > minMergeSize || mergeDocs + segmentDocs > maxMergeDocs) {
+              break;
+            }
+
+            mergeSize += segmentSize;
+            mergeDocs += segmentDocs;
+            end++;
+          }
+        }
+
         if (anyMerging || end - start <= 1) {
           // skip: there is an ongoing merge at the current level or the computed merge has a single
           // segment and this merge policy doesn't do singleton merges

lucene/core/src/test/org/apache/lucene/index/TestLogMergePolicy.java

@@ -50,8 +50,13 @@ protected void assertSegmentInfos(MergePolicy policy, SegmentInfos infos) throws
   @Override
   protected void assertMerge(MergePolicy policy, MergeSpecification merge) throws IOException {
     LogMergePolicy lmp = (LogMergePolicy) policy;
+    MergeContext mockMergeContext = new MockMergeContext(SegmentCommitInfo::getDelCount);
     for (OneMerge oneMerge : merge.merges) {
-      assertTrue(oneMerge.segments.size() <= lmp.getMergeFactor());
+      long mergeSize = 0;
+      for (SegmentCommitInfo info : oneMerge.segments) {
+        mergeSize += lmp.size(info, mockMergeContext);
+      }
+      assertTrue(mergeSize < lmp.minMergeSize || oneMerge.segments.size() <= lmp.getMergeFactor());
     }
   }