@Test public void testMergeWithNoSegments() throws Exception { CommitMitigatingTieredMergePolicy mergePolicy = new CommitMitigatingTieredMergePolicy(); SegmentInfos infos = new SegmentInfos(); MergePolicy.MergeSpecification merges = mergePolicy.findMerges(MergeTrigger.SEGMENT_FLUSH, infos); assertNull(merges); }
@Override public MergeSpecification findMerges(MergeTrigger mergeTrigger, SegmentInfos infos) throws IOException { int segmentSize = infos.size(); timeSeriesCount++; if (timeSeriesCount % timeSeriesLength == 0) { avgCommitRateDocs = 0d; avgCommitRateMB = 0d; avgSegs = 0d; } avgSegs = singleExpSmoothing(segmentSize, avgSegs); log.debug("segments: current {}, average {}", segmentSize, avgSegs); if (verbose()) { message("findMerges: " + segmentSize + " segments, " + avgSegs + " average"); } if (segmentSize == 0) { return null; } if (segmentSize > maxNoOfSegsForMitigation) { if (avgCommitRateDocs > maxCommitRateDocs) { double v = singleExpSmoothing(avgCommitRateDocs, maxCommitRateDocs); log.debug("adjusting maxCommitRateDocs from {} to {}", maxCommitRateDocs, v); maxCommitRateDocs = v; } if (avgCommitRateMB > maxCommitRateMB) { double v = singleExpSmoothing(avgCommitRateMB, maxCommitRateMB); log.debug("adjusting maxCommitRateMB from {} to {}", maxCommitRateMB, v); maxCommitRateMB = v; } } long now = System.currentTimeMillis(); double timeDelta = (now / 1000d) - (time / 1000d); double commitRate = Math.abs(docCount - infos.totalDocCount()) / timeDelta; time = now; avgCommitRateDocs = singleExpSmoothing(commitRate, avgCommitRateDocs); log.debug("commit rate: current {}, average {}, max {} docs/sec", commitRate, avgCommitRateDocs, maxCommitRateDocs); docCount = infos.totalDocCount(); if (verbose()) { message(commitRate + "doc/s (max: " + maxCommitRateDocs + ", avg: " + avgCommitRateDocs + " doc/s)"); } if (commitRate > maxCommitRateDocs && segmentSize < maxNoOfSegsForMitigation) { log.debug("mitigation due to {} > {} docs/sec and segments {} < {})", commitRate, maxCommitRateDocs, segmentSize, maxNoOfSegsForMitigation); return null; } final Collection<SegmentCommitInfo> merging = writer.get().getMergingSegments(); final Collection<SegmentCommitInfo> toBeMerged = new HashSet<SegmentCommitInfo>(); final List<SegmentCommitInfo> infosSorted = new ArrayList<SegmentCommitInfo>(infos.asList()); Collections.sort(infosSorted, new SegmentByteSizeDescending()); long totIndexBytes = 0; long minSegmentBytes = Long.MAX_VALUE; for (SegmentCommitInfo info : infosSorted) { final long segBytes = size(info); if (verbose()) { String extra = merging.contains(info) ? " [merging]" : ""; if (segBytes >= maxMergedSegmentBytes / 2.0) { extra += " [skip: too large]"; } else if (segBytes < floorSegmentBytes) { extra += " [floored]"; } message(" seg=" + writer.get().segString(info) + " size=" + String.format(Locale.ROOT, "%.3f", segBytes / 1024 / 1024.) + " MB" + extra); } minSegmentBytes = Math.min(segBytes, minSegmentBytes); totIndexBytes += segBytes; } int tooBigCount = 0; while (tooBigCount < infosSorted.size() && size(infosSorted.get(tooBigCount)) >= maxMergedSegmentBytes / 2.0) { totIndexBytes -= size(infosSorted.get(tooBigCount)); tooBigCount++; } minSegmentBytes = floorSize(minSegmentBytes); long levelSize = minSegmentBytes; long bytesLeft = totIndexBytes; double allowedSegCount = 0; while (true) { final double segCountLevel = bytesLeft / (double) levelSize; if (segCountLevel < segsPerTier) { allowedSegCount += Math.ceil(segCountLevel); break; } allowedSegCount += segsPerTier; bytesLeft -= segsPerTier * levelSize; levelSize *= maxMergeAtOnce; } int allowedSegCountInt = (int) allowedSegCount; MergeSpecification spec = null; while (true) { long mergingBytes = 0; double idxBytes = 0; final List<SegmentCommitInfo> eligible = new ArrayList<SegmentCommitInfo>(); for (int idx = tooBigCount; idx < infosSorted.size(); idx++) { final SegmentCommitInfo info = infosSorted.get(idx); if (merging.contains(info)) { mergingBytes += info.sizeInBytes(); } else if (!toBeMerged.contains(info)) { eligible.add(info); } idxBytes += info.sizeInBytes(); } idxBytes /= 1024d * 1024d; final boolean maxMergeIsRunning = mergingBytes >= maxMergedSegmentBytes; if (verbose()) { message(" allowedSegmentCount=" + allowedSegCountInt + " vs count=" + infosSorted.size() + " (eligible count=" + eligible.size() + ") tooBigCount=" + tooBigCount); } if (eligible.size() == 0) { return spec; } double bytes = idxBytes - this.mb; double mbRate = bytes / timeDelta; avgCommitRateMB = singleExpSmoothing(mbRate, avgCommitRateMB); log.debug("commit rate: current {}, average {}, max {} MB/sec", mbRate, avgCommitRateMB, maxCommitRateMB); if (verbose()) { message(mbRate + "mb/s (max: " + maxCommitRateMB + ", avg: " + avgCommitRateMB + " MB/s)"); } this.mb = idxBytes; if (mbRate > maxCommitRateMB && segmentSize < maxNoOfSegsForMitigation) { log.debug("mitigation due to {} > {} MB/sec and segments {} < {})", mbRate, maxCommitRateMB, segmentSize, maxNoOfSegsForMitigation); return null; } if (eligible.size() >= allowedSegCountInt) { MergeScore bestScore = null; List<SegmentCommitInfo> best = null; boolean bestTooLarge = false; long bestMergeBytes = 0; for (int startIdx = 0; startIdx <= eligible.size() - maxMergeAtOnce; startIdx++) { long totAfterMergeBytes = 0; final List<SegmentCommitInfo> candidate = new ArrayList<SegmentCommitInfo>(); boolean hitTooLarge = false; for (int idx = startIdx; idx < eligible.size() && candidate.size() < maxMergeAtOnce; idx++) { final SegmentCommitInfo info = eligible.get(idx); final long segBytes = size(info); if (totAfterMergeBytes + segBytes > maxMergedSegmentBytes) { hitTooLarge = true; continue; } candidate.add(info); totAfterMergeBytes += segBytes; } final MergeScore score = score(candidate, hitTooLarge, mergingBytes); if (verbose()) { message(" maybe=" + writer.get().segString(candidate) + " score=" + score.getScore() + " " + score.getExplanation() + " tooLarge=" + hitTooLarge + " size=" + String.format(Locale.ROOT, "%.3f MB", totAfterMergeBytes / 1024. / 1024.)); } if ((bestScore == null || score.getScore() < bestScore.getScore()) && (!hitTooLarge || !maxMergeIsRunning)) { best = candidate; bestScore = score; bestTooLarge = hitTooLarge; bestMergeBytes = totAfterMergeBytes; } } if (best != null) { if (spec == null) { spec = new MergeSpecification(); } final OneMerge merge = new OneMerge(best); spec.add(merge); for (SegmentCommitInfo info : merge.segments) { toBeMerged.add(info); } if (verbose()) { message(" add merge=" + writer.get().segString(merge.segments) + " size=" + String.format(Locale.ROOT, "%.3f MB", bestMergeBytes / 1024. / 1024.) + " score=" + String.format(Locale.ROOT, "%.3f", bestScore.getScore()) + " " + bestScore.getExplanation() + (bestTooLarge ? " [max merge]" : "")); } } else { return spec; } } else { return spec; } } }
CommitMitigatingTieredMergePolicy extends MergePolicy { @Override public MergeSpecification findMerges(MergeTrigger mergeTrigger, SegmentInfos infos) throws IOException { int segmentSize = infos.size(); timeSeriesCount++; if (timeSeriesCount % timeSeriesLength == 0) { avgCommitRateDocs = 0d; avgCommitRateMB = 0d; avgSegs = 0d; } avgSegs = singleExpSmoothing(segmentSize, avgSegs); log.debug("segments: current {}, average {}", segmentSize, avgSegs); if (verbose()) { message("findMerges: " + segmentSize + " segments, " + avgSegs + " average"); } if (segmentSize == 0) { return null; } if (segmentSize > maxNoOfSegsForMitigation) { if (avgCommitRateDocs > maxCommitRateDocs) { double v = singleExpSmoothing(avgCommitRateDocs, maxCommitRateDocs); log.debug("adjusting maxCommitRateDocs from {} to {}", maxCommitRateDocs, v); maxCommitRateDocs = v; } if (avgCommitRateMB > maxCommitRateMB) { double v = singleExpSmoothing(avgCommitRateMB, maxCommitRateMB); log.debug("adjusting maxCommitRateMB from {} to {}", maxCommitRateMB, v); maxCommitRateMB = v; } } long now = System.currentTimeMillis(); double timeDelta = (now / 1000d) - (time / 1000d); double commitRate = Math.abs(docCount - infos.totalDocCount()) / timeDelta; time = now; avgCommitRateDocs = singleExpSmoothing(commitRate, avgCommitRateDocs); log.debug("commit rate: current {}, average {}, max {} docs/sec", commitRate, avgCommitRateDocs, maxCommitRateDocs); docCount = infos.totalDocCount(); if (verbose()) { message(commitRate + "doc/s (max: " + maxCommitRateDocs + ", avg: " + avgCommitRateDocs + " doc/s)"); } if (commitRate > maxCommitRateDocs && segmentSize < maxNoOfSegsForMitigation) { log.debug("mitigation due to {} > {} docs/sec and segments {} < {})", commitRate, maxCommitRateDocs, segmentSize, maxNoOfSegsForMitigation); return null; } final Collection<SegmentCommitInfo> merging = writer.get().getMergingSegments(); final Collection<SegmentCommitInfo> toBeMerged = new HashSet<SegmentCommitInfo>(); final List<SegmentCommitInfo> infosSorted = new ArrayList<SegmentCommitInfo>(infos.asList()); Collections.sort(infosSorted, new SegmentByteSizeDescending()); long totIndexBytes = 0; long minSegmentBytes = Long.MAX_VALUE; for (SegmentCommitInfo info : infosSorted) { final long segBytes = size(info); if (verbose()) { String extra = merging.contains(info) ? " [merging]" : ""; if (segBytes >= maxMergedSegmentBytes / 2.0) { extra += " [skip: too large]"; } else if (segBytes < floorSegmentBytes) { extra += " [floored]"; } message(" seg=" + writer.get().segString(info) + " size=" + String.format(Locale.ROOT, "%.3f", segBytes / 1024 / 1024.) + " MB" + extra); } minSegmentBytes = Math.min(segBytes, minSegmentBytes); totIndexBytes += segBytes; } int tooBigCount = 0; while (tooBigCount < infosSorted.size() && size(infosSorted.get(tooBigCount)) >= maxMergedSegmentBytes / 2.0) { totIndexBytes -= size(infosSorted.get(tooBigCount)); tooBigCount++; } minSegmentBytes = floorSize(minSegmentBytes); long levelSize = minSegmentBytes; long bytesLeft = totIndexBytes; double allowedSegCount = 0; while (true) { final double segCountLevel = bytesLeft / (double) levelSize; if (segCountLevel < segsPerTier) { allowedSegCount += Math.ceil(segCountLevel); break; } allowedSegCount += segsPerTier; bytesLeft -= segsPerTier * levelSize; levelSize *= maxMergeAtOnce; } int allowedSegCountInt = (int) allowedSegCount; MergeSpecification spec = null; while (true) { long mergingBytes = 0; double idxBytes = 0; final List<SegmentCommitInfo> eligible = new ArrayList<SegmentCommitInfo>(); for (int idx = tooBigCount; idx < infosSorted.size(); idx++) { final SegmentCommitInfo info = infosSorted.get(idx); if (merging.contains(info)) { mergingBytes += info.sizeInBytes(); } else if (!toBeMerged.contains(info)) { eligible.add(info); } idxBytes += info.sizeInBytes(); } idxBytes /= 1024d * 1024d; final boolean maxMergeIsRunning = mergingBytes >= maxMergedSegmentBytes; if (verbose()) { message(" allowedSegmentCount=" + allowedSegCountInt + " vs count=" + infosSorted.size() + " (eligible count=" + eligible.size() + ") tooBigCount=" + tooBigCount); } if (eligible.size() == 0) { return spec; } double bytes = idxBytes - this.mb; double mbRate = bytes / timeDelta; avgCommitRateMB = singleExpSmoothing(mbRate, avgCommitRateMB); log.debug("commit rate: current {}, average {}, max {} MB/sec", mbRate, avgCommitRateMB, maxCommitRateMB); if (verbose()) { message(mbRate + "mb/s (max: " + maxCommitRateMB + ", avg: " + avgCommitRateMB + " MB/s)"); } this.mb = idxBytes; if (mbRate > maxCommitRateMB && segmentSize < maxNoOfSegsForMitigation) { log.debug("mitigation due to {} > {} MB/sec and segments {} < {})", mbRate, maxCommitRateMB, segmentSize, maxNoOfSegsForMitigation); return null; } if (eligible.size() >= allowedSegCountInt) { MergeScore bestScore = null; List<SegmentCommitInfo> best = null; boolean bestTooLarge = false; long bestMergeBytes = 0; for (int startIdx = 0; startIdx <= eligible.size() - maxMergeAtOnce; startIdx++) { long totAfterMergeBytes = 0; final List<SegmentCommitInfo> candidate = new ArrayList<SegmentCommitInfo>(); boolean hitTooLarge = false; for (int idx = startIdx; idx < eligible.size() && candidate.size() < maxMergeAtOnce; idx++) { final SegmentCommitInfo info = eligible.get(idx); final long segBytes = size(info); if (totAfterMergeBytes + segBytes > maxMergedSegmentBytes) { hitTooLarge = true; continue; } candidate.add(info); totAfterMergeBytes += segBytes; } final MergeScore score = score(candidate, hitTooLarge, mergingBytes); if (verbose()) { message(" maybe=" + writer.get().segString(candidate) + " score=" + score.getScore() + " " + score.getExplanation() + " tooLarge=" + hitTooLarge + " size=" + String.format(Locale.ROOT, "%.3f MB", totAfterMergeBytes / 1024. / 1024.)); } if ((bestScore == null || score.getScore() < bestScore.getScore()) && (!hitTooLarge || !maxMergeIsRunning)) { best = candidate; bestScore = score; bestTooLarge = hitTooLarge; bestMergeBytes = totAfterMergeBytes; } } if (best != null) { if (spec == null) { spec = new MergeSpecification(); } final OneMerge merge = new OneMerge(best); spec.add(merge); for (SegmentCommitInfo info : merge.segments) { toBeMerged.add(info); } if (verbose()) { message(" add merge=" + writer.get().segString(merge.segments) + " size=" + String.format(Locale.ROOT, "%.3f MB", bestMergeBytes / 1024. / 1024.) + " score=" + String.format(Locale.ROOT, "%.3f", bestScore.getScore()) + " " + bestScore.getExplanation() + (bestTooLarge ? " [max merge]" : "")); } } else { return spec; } } else { return spec; } } } }
CommitMitigatingTieredMergePolicy extends MergePolicy { @Override public MergeSpecification findMerges(MergeTrigger mergeTrigger, SegmentInfos infos) throws IOException { int segmentSize = infos.size(); timeSeriesCount++; if (timeSeriesCount % timeSeriesLength == 0) { avgCommitRateDocs = 0d; avgCommitRateMB = 0d; avgSegs = 0d; } avgSegs = singleExpSmoothing(segmentSize, avgSegs); log.debug("segments: current {}, average {}", segmentSize, avgSegs); if (verbose()) { message("findMerges: " + segmentSize + " segments, " + avgSegs + " average"); } if (segmentSize == 0) { return null; } if (segmentSize > maxNoOfSegsForMitigation) { if (avgCommitRateDocs > maxCommitRateDocs) { double v = singleExpSmoothing(avgCommitRateDocs, maxCommitRateDocs); log.debug("adjusting maxCommitRateDocs from {} to {}", maxCommitRateDocs, v); maxCommitRateDocs = v; } if (avgCommitRateMB > maxCommitRateMB) { double v = singleExpSmoothing(avgCommitRateMB, maxCommitRateMB); log.debug("adjusting maxCommitRateMB from {} to {}", maxCommitRateMB, v); maxCommitRateMB = v; } } long now = System.currentTimeMillis(); double timeDelta = (now / 1000d) - (time / 1000d); double commitRate = Math.abs(docCount - infos.totalDocCount()) / timeDelta; time = now; avgCommitRateDocs = singleExpSmoothing(commitRate, avgCommitRateDocs); log.debug("commit rate: current {}, average {}, max {} docs/sec", commitRate, avgCommitRateDocs, maxCommitRateDocs); docCount = infos.totalDocCount(); if (verbose()) { message(commitRate + "doc/s (max: " + maxCommitRateDocs + ", avg: " + avgCommitRateDocs + " doc/s)"); } if (commitRate > maxCommitRateDocs && segmentSize < maxNoOfSegsForMitigation) { log.debug("mitigation due to {} > {} docs/sec and segments {} < {})", commitRate, maxCommitRateDocs, segmentSize, maxNoOfSegsForMitigation); return null; } final Collection<SegmentCommitInfo> merging = writer.get().getMergingSegments(); final Collection<SegmentCommitInfo> toBeMerged = new HashSet<SegmentCommitInfo>(); final List<SegmentCommitInfo> infosSorted = new ArrayList<SegmentCommitInfo>(infos.asList()); Collections.sort(infosSorted, new SegmentByteSizeDescending()); long totIndexBytes = 0; long minSegmentBytes = Long.MAX_VALUE; for (SegmentCommitInfo info : infosSorted) { final long segBytes = size(info); if (verbose()) { String extra = merging.contains(info) ? " [merging]" : ""; if (segBytes >= maxMergedSegmentBytes / 2.0) { extra += " [skip: too large]"; } else if (segBytes < floorSegmentBytes) { extra += " [floored]"; } message(" seg=" + writer.get().segString(info) + " size=" + String.format(Locale.ROOT, "%.3f", segBytes / 1024 / 1024.) + " MB" + extra); } minSegmentBytes = Math.min(segBytes, minSegmentBytes); totIndexBytes += segBytes; } int tooBigCount = 0; while (tooBigCount < infosSorted.size() && size(infosSorted.get(tooBigCount)) >= maxMergedSegmentBytes / 2.0) { totIndexBytes -= size(infosSorted.get(tooBigCount)); tooBigCount++; } minSegmentBytes = floorSize(minSegmentBytes); long levelSize = minSegmentBytes; long bytesLeft = totIndexBytes; double allowedSegCount = 0; while (true) { final double segCountLevel = bytesLeft / (double) levelSize; if (segCountLevel < segsPerTier) { allowedSegCount += Math.ceil(segCountLevel); break; } allowedSegCount += segsPerTier; bytesLeft -= segsPerTier * levelSize; levelSize *= maxMergeAtOnce; } int allowedSegCountInt = (int) allowedSegCount; MergeSpecification spec = null; while (true) { long mergingBytes = 0; double idxBytes = 0; final List<SegmentCommitInfo> eligible = new ArrayList<SegmentCommitInfo>(); for (int idx = tooBigCount; idx < infosSorted.size(); idx++) { final SegmentCommitInfo info = infosSorted.get(idx); if (merging.contains(info)) { mergingBytes += info.sizeInBytes(); } else if (!toBeMerged.contains(info)) { eligible.add(info); } idxBytes += info.sizeInBytes(); } idxBytes /= 1024d * 1024d; final boolean maxMergeIsRunning = mergingBytes >= maxMergedSegmentBytes; if (verbose()) { message(" allowedSegmentCount=" + allowedSegCountInt + " vs count=" + infosSorted.size() + " (eligible count=" + eligible.size() + ") tooBigCount=" + tooBigCount); } if (eligible.size() == 0) { return spec; } double bytes = idxBytes - this.mb; double mbRate = bytes / timeDelta; avgCommitRateMB = singleExpSmoothing(mbRate, avgCommitRateMB); log.debug("commit rate: current {}, average {}, max {} MB/sec", mbRate, avgCommitRateMB, maxCommitRateMB); if (verbose()) { message(mbRate + "mb/s (max: " + maxCommitRateMB + ", avg: " + avgCommitRateMB + " MB/s)"); } this.mb = idxBytes; if (mbRate > maxCommitRateMB && segmentSize < maxNoOfSegsForMitigation) { log.debug("mitigation due to {} > {} MB/sec and segments {} < {})", mbRate, maxCommitRateMB, segmentSize, maxNoOfSegsForMitigation); return null; } if (eligible.size() >= allowedSegCountInt) { MergeScore bestScore = null; List<SegmentCommitInfo> best = null; boolean bestTooLarge = false; long bestMergeBytes = 0; for (int startIdx = 0; startIdx <= eligible.size() - maxMergeAtOnce; startIdx++) { long totAfterMergeBytes = 0; final List<SegmentCommitInfo> candidate = new ArrayList<SegmentCommitInfo>(); boolean hitTooLarge = false; for (int idx = startIdx; idx < eligible.size() && candidate.size() < maxMergeAtOnce; idx++) { final SegmentCommitInfo info = eligible.get(idx); final long segBytes = size(info); if (totAfterMergeBytes + segBytes > maxMergedSegmentBytes) { hitTooLarge = true; continue; } candidate.add(info); totAfterMergeBytes += segBytes; } final MergeScore score = score(candidate, hitTooLarge, mergingBytes); if (verbose()) { message(" maybe=" + writer.get().segString(candidate) + " score=" + score.getScore() + " " + score.getExplanation() + " tooLarge=" + hitTooLarge + " size=" + String.format(Locale.ROOT, "%.3f MB", totAfterMergeBytes / 1024. / 1024.)); } if ((bestScore == null || score.getScore() < bestScore.getScore()) && (!hitTooLarge || !maxMergeIsRunning)) { best = candidate; bestScore = score; bestTooLarge = hitTooLarge; bestMergeBytes = totAfterMergeBytes; } } if (best != null) { if (spec == null) { spec = new MergeSpecification(); } final OneMerge merge = new OneMerge(best); spec.add(merge); for (SegmentCommitInfo info : merge.segments) { toBeMerged.add(info); } if (verbose()) { message(" add merge=" + writer.get().segString(merge.segments) + " size=" + String.format(Locale.ROOT, "%.3f MB", bestMergeBytes / 1024. / 1024.) + " score=" + String.format(Locale.ROOT, "%.3f", bestScore.getScore()) + " " + bestScore.getExplanation() + (bestTooLarge ? " [max merge]" : "")); } } else { return spec; } } else { return spec; } } } CommitMitigatingTieredMergePolicy(); }
CommitMitigatingTieredMergePolicy extends MergePolicy { @Override public MergeSpecification findMerges(MergeTrigger mergeTrigger, SegmentInfos infos) throws IOException { int segmentSize = infos.size(); timeSeriesCount++; if (timeSeriesCount % timeSeriesLength == 0) { avgCommitRateDocs = 0d; avgCommitRateMB = 0d; avgSegs = 0d; } avgSegs = singleExpSmoothing(segmentSize, avgSegs); log.debug("segments: current {}, average {}", segmentSize, avgSegs); if (verbose()) { message("findMerges: " + segmentSize + " segments, " + avgSegs + " average"); } if (segmentSize == 0) { return null; } if (segmentSize > maxNoOfSegsForMitigation) { if (avgCommitRateDocs > maxCommitRateDocs) { double v = singleExpSmoothing(avgCommitRateDocs, maxCommitRateDocs); log.debug("adjusting maxCommitRateDocs from {} to {}", maxCommitRateDocs, v); maxCommitRateDocs = v; } if (avgCommitRateMB > maxCommitRateMB) { double v = singleExpSmoothing(avgCommitRateMB, maxCommitRateMB); log.debug("adjusting maxCommitRateMB from {} to {}", maxCommitRateMB, v); maxCommitRateMB = v; } } long now = System.currentTimeMillis(); double timeDelta = (now / 1000d) - (time / 1000d); double commitRate = Math.abs(docCount - infos.totalDocCount()) / timeDelta; time = now; avgCommitRateDocs = singleExpSmoothing(commitRate, avgCommitRateDocs); log.debug("commit rate: current {}, average {}, max {} docs/sec", commitRate, avgCommitRateDocs, maxCommitRateDocs); docCount = infos.totalDocCount(); if (verbose()) { message(commitRate + "doc/s (max: " + maxCommitRateDocs + ", avg: " + avgCommitRateDocs + " doc/s)"); } if (commitRate > maxCommitRateDocs && segmentSize < maxNoOfSegsForMitigation) { log.debug("mitigation due to {} > {} docs/sec and segments {} < {})", commitRate, maxCommitRateDocs, segmentSize, maxNoOfSegsForMitigation); return null; } final Collection<SegmentCommitInfo> merging = writer.get().getMergingSegments(); final Collection<SegmentCommitInfo> toBeMerged = new HashSet<SegmentCommitInfo>(); final List<SegmentCommitInfo> infosSorted = new ArrayList<SegmentCommitInfo>(infos.asList()); Collections.sort(infosSorted, new SegmentByteSizeDescending()); long totIndexBytes = 0; long minSegmentBytes = Long.MAX_VALUE; for (SegmentCommitInfo info : infosSorted) { final long segBytes = size(info); if (verbose()) { String extra = merging.contains(info) ? " [merging]" : ""; if (segBytes >= maxMergedSegmentBytes / 2.0) { extra += " [skip: too large]"; } else if (segBytes < floorSegmentBytes) { extra += " [floored]"; } message(" seg=" + writer.get().segString(info) + " size=" + String.format(Locale.ROOT, "%.3f", segBytes / 1024 / 1024.) + " MB" + extra); } minSegmentBytes = Math.min(segBytes, minSegmentBytes); totIndexBytes += segBytes; } int tooBigCount = 0; while (tooBigCount < infosSorted.size() && size(infosSorted.get(tooBigCount)) >= maxMergedSegmentBytes / 2.0) { totIndexBytes -= size(infosSorted.get(tooBigCount)); tooBigCount++; } minSegmentBytes = floorSize(minSegmentBytes); long levelSize = minSegmentBytes; long bytesLeft = totIndexBytes; double allowedSegCount = 0; while (true) { final double segCountLevel = bytesLeft / (double) levelSize; if (segCountLevel < segsPerTier) { allowedSegCount += Math.ceil(segCountLevel); break; } allowedSegCount += segsPerTier; bytesLeft -= segsPerTier * levelSize; levelSize *= maxMergeAtOnce; } int allowedSegCountInt = (int) allowedSegCount; MergeSpecification spec = null; while (true) { long mergingBytes = 0; double idxBytes = 0; final List<SegmentCommitInfo> eligible = new ArrayList<SegmentCommitInfo>(); for (int idx = tooBigCount; idx < infosSorted.size(); idx++) { final SegmentCommitInfo info = infosSorted.get(idx); if (merging.contains(info)) { mergingBytes += info.sizeInBytes(); } else if (!toBeMerged.contains(info)) { eligible.add(info); } idxBytes += info.sizeInBytes(); } idxBytes /= 1024d * 1024d; final boolean maxMergeIsRunning = mergingBytes >= maxMergedSegmentBytes; if (verbose()) { message(" allowedSegmentCount=" + allowedSegCountInt + " vs count=" + infosSorted.size() + " (eligible count=" + eligible.size() + ") tooBigCount=" + tooBigCount); } if (eligible.size() == 0) { return spec; } double bytes = idxBytes - this.mb; double mbRate = bytes / timeDelta; avgCommitRateMB = singleExpSmoothing(mbRate, avgCommitRateMB); log.debug("commit rate: current {}, average {}, max {} MB/sec", mbRate, avgCommitRateMB, maxCommitRateMB); if (verbose()) { message(mbRate + "mb/s (max: " + maxCommitRateMB + ", avg: " + avgCommitRateMB + " MB/s)"); } this.mb = idxBytes; if (mbRate > maxCommitRateMB && segmentSize < maxNoOfSegsForMitigation) { log.debug("mitigation due to {} > {} MB/sec and segments {} < {})", mbRate, maxCommitRateMB, segmentSize, maxNoOfSegsForMitigation); return null; } if (eligible.size() >= allowedSegCountInt) { MergeScore bestScore = null; List<SegmentCommitInfo> best = null; boolean bestTooLarge = false; long bestMergeBytes = 0; for (int startIdx = 0; startIdx <= eligible.size() - maxMergeAtOnce; startIdx++) { long totAfterMergeBytes = 0; final List<SegmentCommitInfo> candidate = new ArrayList<SegmentCommitInfo>(); boolean hitTooLarge = false; for (int idx = startIdx; idx < eligible.size() && candidate.size() < maxMergeAtOnce; idx++) { final SegmentCommitInfo info = eligible.get(idx); final long segBytes = size(info); if (totAfterMergeBytes + segBytes > maxMergedSegmentBytes) { hitTooLarge = true; continue; } candidate.add(info); totAfterMergeBytes += segBytes; } final MergeScore score = score(candidate, hitTooLarge, mergingBytes); if (verbose()) { message(" maybe=" + writer.get().segString(candidate) + " score=" + score.getScore() + " " + score.getExplanation() + " tooLarge=" + hitTooLarge + " size=" + String.format(Locale.ROOT, "%.3f MB", totAfterMergeBytes / 1024. / 1024.)); } if ((bestScore == null || score.getScore() < bestScore.getScore()) && (!hitTooLarge || !maxMergeIsRunning)) { best = candidate; bestScore = score; bestTooLarge = hitTooLarge; bestMergeBytes = totAfterMergeBytes; } } if (best != null) { if (spec == null) { spec = new MergeSpecification(); } final OneMerge merge = new OneMerge(best); spec.add(merge); for (SegmentCommitInfo info : merge.segments) { toBeMerged.add(info); } if (verbose()) { message(" add merge=" + writer.get().segString(merge.segments) + " size=" + String.format(Locale.ROOT, "%.3f MB", bestMergeBytes / 1024. / 1024.) + " score=" + String.format(Locale.ROOT, "%.3f", bestScore.getScore()) + " " + bestScore.getExplanation() + (bestTooLarge ? " [max merge]" : "")); } } else { return spec; } } else { return spec; } } } CommitMitigatingTieredMergePolicy(); CommitMitigatingTieredMergePolicy setMaxMergeAtOnce(int v); CommitMitigatingTieredMergePolicy setMaxNoOfSegsForMitigation(int maxNoOfSegsForMitigation); CommitMitigatingTieredMergePolicy setMaxCommitRateDocs(double maxCommitRate); int getMaxMergeAtOnce(); CommitMitigatingTieredMergePolicy setMaxMergeAtOnceExplicit(int v); int getMaxMergeAtOnceExplicit(); CommitMitigatingTieredMergePolicy setMaxMergedSegmentMB(double v); double getMaxMergedSegmentMB(); CommitMitigatingTieredMergePolicy setReclaimDeletesWeight(double v); double getReclaimDeletesWeight(); CommitMitigatingTieredMergePolicy setFloorSegmentMB(double v); double getFloorSegmentMB(); CommitMitigatingTieredMergePolicy setForceMergeDeletesPctAllowed(double v); double getForceMergeDeletesPctAllowed(); CommitMitigatingTieredMergePolicy setSegmentsPerTier(double v); double getSegmentsPerTier(); CommitMitigatingTieredMergePolicy setMaxCommitRateMB(int maxCommitRateMB); @Override MergeSpecification findMerges(MergeTrigger mergeTrigger, SegmentInfos infos); @Override MergeSpecification findForcedMerges(SegmentInfos infos, int maxSegmentCount, Map<SegmentCommitInfo, Boolean> segmentsToMerge); @Override MergeSpecification findForcedDeletesMerges(SegmentInfos infos); @Override void close(); @Override String toString(); }
CommitMitigatingTieredMergePolicy extends MergePolicy { @Override public MergeSpecification findMerges(MergeTrigger mergeTrigger, SegmentInfos infos) throws IOException { int segmentSize = infos.size(); timeSeriesCount++; if (timeSeriesCount % timeSeriesLength == 0) { avgCommitRateDocs = 0d; avgCommitRateMB = 0d; avgSegs = 0d; } avgSegs = singleExpSmoothing(segmentSize, avgSegs); log.debug("segments: current {}, average {}", segmentSize, avgSegs); if (verbose()) { message("findMerges: " + segmentSize + " segments, " + avgSegs + " average"); } if (segmentSize == 0) { return null; } if (segmentSize > maxNoOfSegsForMitigation) { if (avgCommitRateDocs > maxCommitRateDocs) { double v = singleExpSmoothing(avgCommitRateDocs, maxCommitRateDocs); log.debug("adjusting maxCommitRateDocs from {} to {}", maxCommitRateDocs, v); maxCommitRateDocs = v; } if (avgCommitRateMB > maxCommitRateMB) { double v = singleExpSmoothing(avgCommitRateMB, maxCommitRateMB); log.debug("adjusting maxCommitRateMB from {} to {}", maxCommitRateMB, v); maxCommitRateMB = v; } } long now = System.currentTimeMillis(); double timeDelta = (now / 1000d) - (time / 1000d); double commitRate = Math.abs(docCount - infos.totalDocCount()) / timeDelta; time = now; avgCommitRateDocs = singleExpSmoothing(commitRate, avgCommitRateDocs); log.debug("commit rate: current {}, average {}, max {} docs/sec", commitRate, avgCommitRateDocs, maxCommitRateDocs); docCount = infos.totalDocCount(); if (verbose()) { message(commitRate + "doc/s (max: " + maxCommitRateDocs + ", avg: " + avgCommitRateDocs + " doc/s)"); } if (commitRate > maxCommitRateDocs && segmentSize < maxNoOfSegsForMitigation) { log.debug("mitigation due to {} > {} docs/sec and segments {} < {})", commitRate, maxCommitRateDocs, segmentSize, maxNoOfSegsForMitigation); return null; } final Collection<SegmentCommitInfo> merging = writer.get().getMergingSegments(); final Collection<SegmentCommitInfo> toBeMerged = new HashSet<SegmentCommitInfo>(); final List<SegmentCommitInfo> infosSorted = new ArrayList<SegmentCommitInfo>(infos.asList()); Collections.sort(infosSorted, new SegmentByteSizeDescending()); long totIndexBytes = 0; long minSegmentBytes = Long.MAX_VALUE; for (SegmentCommitInfo info : infosSorted) { final long segBytes = size(info); if (verbose()) { String extra = merging.contains(info) ? " [merging]" : ""; if (segBytes >= maxMergedSegmentBytes / 2.0) { extra += " [skip: too large]"; } else if (segBytes < floorSegmentBytes) { extra += " [floored]"; } message(" seg=" + writer.get().segString(info) + " size=" + String.format(Locale.ROOT, "%.3f", segBytes / 1024 / 1024.) + " MB" + extra); } minSegmentBytes = Math.min(segBytes, minSegmentBytes); totIndexBytes += segBytes; } int tooBigCount = 0; while (tooBigCount < infosSorted.size() && size(infosSorted.get(tooBigCount)) >= maxMergedSegmentBytes / 2.0) { totIndexBytes -= size(infosSorted.get(tooBigCount)); tooBigCount++; } minSegmentBytes = floorSize(minSegmentBytes); long levelSize = minSegmentBytes; long bytesLeft = totIndexBytes; double allowedSegCount = 0; while (true) { final double segCountLevel = bytesLeft / (double) levelSize; if (segCountLevel < segsPerTier) { allowedSegCount += Math.ceil(segCountLevel); break; } allowedSegCount += segsPerTier; bytesLeft -= segsPerTier * levelSize; levelSize *= maxMergeAtOnce; } int allowedSegCountInt = (int) allowedSegCount; MergeSpecification spec = null; while (true) { long mergingBytes = 0; double idxBytes = 0; final List<SegmentCommitInfo> eligible = new ArrayList<SegmentCommitInfo>(); for (int idx = tooBigCount; idx < infosSorted.size(); idx++) { final SegmentCommitInfo info = infosSorted.get(idx); if (merging.contains(info)) { mergingBytes += info.sizeInBytes(); } else if (!toBeMerged.contains(info)) { eligible.add(info); } idxBytes += info.sizeInBytes(); } idxBytes /= 1024d * 1024d; final boolean maxMergeIsRunning = mergingBytes >= maxMergedSegmentBytes; if (verbose()) { message(" allowedSegmentCount=" + allowedSegCountInt + " vs count=" + infosSorted.size() + " (eligible count=" + eligible.size() + ") tooBigCount=" + tooBigCount); } if (eligible.size() == 0) { return spec; } double bytes = idxBytes - this.mb; double mbRate = bytes / timeDelta; avgCommitRateMB = singleExpSmoothing(mbRate, avgCommitRateMB); log.debug("commit rate: current {}, average {}, max {} MB/sec", mbRate, avgCommitRateMB, maxCommitRateMB); if (verbose()) { message(mbRate + "mb/s (max: " + maxCommitRateMB + ", avg: " + avgCommitRateMB + " MB/s)"); } this.mb = idxBytes; if (mbRate > maxCommitRateMB && segmentSize < maxNoOfSegsForMitigation) { log.debug("mitigation due to {} > {} MB/sec and segments {} < {})", mbRate, maxCommitRateMB, segmentSize, maxNoOfSegsForMitigation); return null; } if (eligible.size() >= allowedSegCountInt) { MergeScore bestScore = null; List<SegmentCommitInfo> best = null; boolean bestTooLarge = false; long bestMergeBytes = 0; for (int startIdx = 0; startIdx <= eligible.size() - maxMergeAtOnce; startIdx++) { long totAfterMergeBytes = 0; final List<SegmentCommitInfo> candidate = new ArrayList<SegmentCommitInfo>(); boolean hitTooLarge = false; for (int idx = startIdx; idx < eligible.size() && candidate.size() < maxMergeAtOnce; idx++) { final SegmentCommitInfo info = eligible.get(idx); final long segBytes = size(info); if (totAfterMergeBytes + segBytes > maxMergedSegmentBytes) { hitTooLarge = true; continue; } candidate.add(info); totAfterMergeBytes += segBytes; } final MergeScore score = score(candidate, hitTooLarge, mergingBytes); if (verbose()) { message(" maybe=" + writer.get().segString(candidate) + " score=" + score.getScore() + " " + score.getExplanation() + " tooLarge=" + hitTooLarge + " size=" + String.format(Locale.ROOT, "%.3f MB", totAfterMergeBytes / 1024. / 1024.)); } if ((bestScore == null || score.getScore() < bestScore.getScore()) && (!hitTooLarge || !maxMergeIsRunning)) { best = candidate; bestScore = score; bestTooLarge = hitTooLarge; bestMergeBytes = totAfterMergeBytes; } } if (best != null) { if (spec == null) { spec = new MergeSpecification(); } final OneMerge merge = new OneMerge(best); spec.add(merge); for (SegmentCommitInfo info : merge.segments) { toBeMerged.add(info); } if (verbose()) { message(" add merge=" + writer.get().segString(merge.segments) + " size=" + String.format(Locale.ROOT, "%.3f MB", bestMergeBytes / 1024. / 1024.) + " score=" + String.format(Locale.ROOT, "%.3f", bestScore.getScore()) + " " + bestScore.getExplanation() + (bestTooLarge ? " [max merge]" : "")); } } else { return spec; } } else { return spec; } } } CommitMitigatingTieredMergePolicy(); CommitMitigatingTieredMergePolicy setMaxMergeAtOnce(int v); CommitMitigatingTieredMergePolicy setMaxNoOfSegsForMitigation(int maxNoOfSegsForMitigation); CommitMitigatingTieredMergePolicy setMaxCommitRateDocs(double maxCommitRate); int getMaxMergeAtOnce(); CommitMitigatingTieredMergePolicy setMaxMergeAtOnceExplicit(int v); int getMaxMergeAtOnceExplicit(); CommitMitigatingTieredMergePolicy setMaxMergedSegmentMB(double v); double getMaxMergedSegmentMB(); CommitMitigatingTieredMergePolicy setReclaimDeletesWeight(double v); double getReclaimDeletesWeight(); CommitMitigatingTieredMergePolicy setFloorSegmentMB(double v); double getFloorSegmentMB(); CommitMitigatingTieredMergePolicy setForceMergeDeletesPctAllowed(double v); double getForceMergeDeletesPctAllowed(); CommitMitigatingTieredMergePolicy setSegmentsPerTier(double v); double getSegmentsPerTier(); CommitMitigatingTieredMergePolicy setMaxCommitRateMB(int maxCommitRateMB); @Override MergeSpecification findMerges(MergeTrigger mergeTrigger, SegmentInfos infos); @Override MergeSpecification findForcedMerges(SegmentInfos infos, int maxSegmentCount, Map<SegmentCommitInfo, Boolean> segmentsToMerge); @Override MergeSpecification findForcedDeletesMerges(SegmentInfos infos); @Override void close(); @Override String toString(); static final double DEFAULT_NO_CFS_RATIO; }