Add new compaction policy to prioritize fragmented intervals

server/src/main/java/org/apache/druid/server/compaction/BaseCandidateSearchPolicy.java

@@ -66,7 +66,6 @@ public final int compareCandidates(CompactionCandidate o1, CompactionCandidate o
   @Override
   public boolean isEligibleForCompaction(
       CompactionCandidate candidate,
-      CompactionStatus currentCompactionStatus,
       CompactionTaskStatus latestTaskStatus
   )
   {

server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java

@@ -137,6 +137,20 @@ public CompactionStatistics getStats()
     return CompactionStatistics.create(totalBytes, numSegments(), numIntervals);
   }
 
+  @Nullable
+  public CompactionStatistics getCompactedStats()
+  {
+    return (currentStatus == null || currentStatus.getCompactedStats() == null)
+           ? null : currentStatus.getCompactedStats();
+  }
+
+  @Nullable
+  public CompactionStatistics getUncompactedStats()
+  {
+    return (currentStatus == null || currentStatus.getUncompactedStats() == null)
+           ? null : currentStatus.getUncompactedStats();
+  }
+
   /**
    * Current compaction status of the time chunk corresponding to this candidate.
    */

server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java

@@ -29,7 +29,8 @@
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 @JsonSubTypes(value = {
     @JsonSubTypes.Type(name = "newestSegmentFirst", value = NewestSegmentFirstPolicy.class),
-    @JsonSubTypes.Type(name = "fixedIntervalOrder", value = FixedIntervalOrderPolicy.class)
+    @JsonSubTypes.Type(name = "fixedIntervalOrder", value = FixedIntervalOrderPolicy.class),
+    @JsonSubTypes.Type(name = "mostFragmentedFirst", value = MostFragmentedIntervalFirstPolicy.class)
 })
 public interface CompactionCandidateSearchPolicy
 {
@@ -50,7 +51,6 @@ public interface CompactionCandidateSearchPolicy
    */
   boolean isEligibleForCompaction(
       CompactionCandidate candidate,
-      CompactionStatus currentCompactionStatus,
       CompactionTaskStatus latestTaskStatus
   );
 }

server/src/main/java/org/apache/druid/server/compaction/CompactionStatistics.java

@@ -65,4 +65,14 @@ public void decrement(CompactionStatistics other)
     numIntervals -= other.getNumIntervals();
     numSegments -= other.getNumSegments();
   }
+
+  @Override
+  public String toString()
+  {
+    return "CompactionStatistics{" +
+           "totalBytes=" + totalBytes +
+           ", numSegments=" + numSegments +
+           ", numIntervals=" + numIntervals +
+           '}';
+  }
 }