Add new compaction policy to prioritize fragmented intervals (#18802)

Author: kfaraz

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

@@ -64,13 +64,12 @@ public final int compareCandidates(CompactionCandidate o1, CompactionCandidate o
   }
 
   @Override
-  public boolean isEligibleForCompaction(
+  public Eligibility checkEligibilityForCompaction(
       CompactionCandidate candidate,
-      CompactionStatus currentCompactionStatus,
       CompactionTaskStatus latestTaskStatus
   )
   {
-    return true;
+    return Eligibility.OK;
   }
 
   /**

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

@@ -21,24 +21,28 @@
 
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.server.coordinator.duty.CompactSegments;
 
+import java.util.Objects;
+
 /**
  * Policy used by {@link CompactSegments} duty to pick segments for compaction.
  */
 @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
 {
   /**
    * Compares between two compaction candidates. Used to determine the
    * order in which segments and intervals should be picked for compaction.
    *
-   * @return A positive value if {@code candidateA} should be picked first, a
-   * negative value if {@code candidateB} should be picked first or zero if the
+   * @return A negative value if {@code candidateA} should be picked first, a
+   * positive value if {@code candidateB} should be picked first or zero if the
    * order does not matter.
    */
   int compareCandidates(CompactionCandidate candidateA, CompactionCandidate candidateB);
@@ -47,10 +51,71 @@ public interface CompactionCandidateSearchPolicy
    * Checks if the given {@link CompactionCandidate} is eligible for compaction
    * in the current iteration. A policy may implement this method to skip
    * compacting intervals or segments that do not fulfil some required criteria.
+   *
+   * @return {@link Eligibility#OK} only if eligible.
    */
-  boolean isEligibleForCompaction(
+  Eligibility checkEligibilityForCompaction(
       CompactionCandidate candidate,
-      CompactionStatus currentCompactionStatus,
       CompactionTaskStatus latestTaskStatus
   );
+
+  /**
+   * Describes the eligibility of an interval for compaction.
+   */
+  class Eligibility
+  {
+    public static final Eligibility OK = new Eligibility(true, null);
+
+    private final boolean eligible;
+    private final String reason;
+
+    private Eligibility(boolean eligible, String reason)
+    {
+      this.eligible = eligible;
+      this.reason = reason;
+    }
+
+    public boolean isEligible()
+    {
+      return eligible;
+    }
+
+    public String getReason()
+    {
+      return reason;
+    }
+
+    public static Eligibility fail(String messageFormat, Object... args)
+    {
+      return new Eligibility(false, StringUtils.format(messageFormat, args));
+    }
+
+    @Override
+    public boolean equals(Object object)
+    {
+      if (this == object) {
+        return true;
+      }
+      if (object == null || getClass() != object.getClass()) {
+        return false;
+      }
+      Eligibility that = (Eligibility) object;
+      return eligible == that.eligible && Objects.equals(reason, that.reason);
+    }
+
+    @Override
+    public int hashCode()
+    {
+      return Objects.hash(eligible, reason);
+    }
+
+    @Override
+    public String toString()
+    {
+      return "Eligibility{" +
+             "eligible=" + eligible +
+             ", reason='" + reason + '\'' +
+             '}';
+    }
+  }
 }

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 +
+           '}';
+  }
 }