MostFragmentedIntervalFirstPolicy: Implement equals. (#18820)

* MostFragmentedIntervalFirstPolicy: Implement equals.

Also hashCode and toString. This is required for correct behavior,
since the policy will not be saved to the database on edit if it
compares equal to the existing, and the base class BaseCandidateSearchPolicy
does have an equals method. So, subclasses such as this one must
override equals and hashCode.

* Remove unused import.

Author: gianm

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

@@ -19,7 +19,9 @@
 
 package org.apache.druid.server.compaction;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.annotations.SubclassesMustOverrideEqualsAndHashCode;
 import org.apache.druid.java.util.common.guava.Comparators;
 
 import javax.annotation.Nullable;
@@ -30,6 +32,7 @@
  * Base implementation of {@link CompactionCandidateSearchPolicy} that can have
  * a {@code priorityDatasource}.
  */
+@SubclassesMustOverrideEqualsAndHashCode
 public abstract class BaseCandidateSearchPolicy implements CompactionCandidateSearchPolicy
 {
   private final String priorityDatasource;
@@ -52,6 +55,7 @@ protected BaseCandidateSearchPolicy(@Nullable String priorityDatasource)
    */
   @Nullable
   @JsonProperty
+  @JsonInclude(JsonInclude.Include.NON_NULL)
   public final String getPriorityDatasource()
   {
     return priorityDatasource;

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

@@ -28,6 +28,7 @@
 
 import javax.annotation.Nullable;
 import java.util.Comparator;
+import java.util.Objects;
 
 /**
  * Experimental {@link CompactionCandidateSearchPolicy} which prioritizes compaction
@@ -111,6 +112,43 @@ protected Comparator<CompactionCandidate> getSegmentComparator()
     return this::compare;
   }
 
+  @Override
+  public boolean equals(Object o)
+  {
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    if (!super.equals(o)) {
+      return false;
+    }
+    MostFragmentedIntervalFirstPolicy policy = (MostFragmentedIntervalFirstPolicy) o;
+    return minUncompactedCount == policy.minUncompactedCount
+           && Objects.equals(minUncompactedBytes, policy.minUncompactedBytes)
+           && Objects.equals(maxAverageUncompactedBytesPerSegment, policy.maxAverageUncompactedBytesPerSegment);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(
+        super.hashCode(),
+        minUncompactedCount,
+        minUncompactedBytes,
+        maxAverageUncompactedBytesPerSegment
+    );
+  }
+
+  @Override
+  public String toString()
+  {
+    return "MostFragmentedIntervalFirstPolicy{" +
+           "minUncompactedCount=" + minUncompactedCount +
+           ", minUncompactedBytes=" + minUncompactedBytes +
+           ", maxAverageUncompactedBytesPerSegment=" + maxAverageUncompactedBytesPerSegment +
+           ", priorityDataSource='" + getPriorityDatasource() + '\'' +
+           '}';
+  }
+
   private int compare(CompactionCandidate candidateA, CompactionCandidate candidateB)
   {
     final double fragmentationDiff

server/src/test/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicyTest.java

@@ -19,13 +19,16 @@
 
 package org.apache.druid.server.compaction;
 
+import nl.jqno.equalsverifier.EqualsVerifier;
+import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.common.HumanReadableBytes;
 import org.apache.druid.segment.TestDataSource;
 import org.apache.druid.server.coordinator.CreateDataSegments;
 import org.apache.druid.timeline.DataSegment;
 import org.junit.Test;
 import org.junit.jupiter.api.Assertions;
 
+import java.io.IOException;
 import java.util.List;
 
 public class MostFragmentedIntervalFirstPolicyTest
@@ -192,6 +195,40 @@ public void test_compareCandidates_returnsZeroIfSegmentCountAndAvgSizeScaleEquiv
     Assertions.assertEquals(0, policy.compareCandidates(candidateB, candidateA));
   }
 
+  @Test
+  public void test_equals()
+  {
+    EqualsVerifier.forClass(MostFragmentedIntervalFirstPolicy.class)
+                  .usingGetClass()
+                  .withIgnoredFields("comparator")
+                  .verify();
+  }
+
+  @Test
+  public void test_serde_allFieldsSet() throws IOException
+  {
+    final MostFragmentedIntervalFirstPolicy policy = new MostFragmentedIntervalFirstPolicy(
+        1,
+        HumanReadableBytes.valueOf(2),
+        HumanReadableBytes.valueOf(3),
+        "foo"
+    );
+    final DefaultObjectMapper mapper = new DefaultObjectMapper();
+    final CompactionCandidateSearchPolicy policy2 =
+        mapper.readValue(mapper.writeValueAsString(policy), CompactionCandidateSearchPolicy.class);
+    Assertions.assertEquals(policy, policy2);
+  }
+
+  @Test
+  public void test_serde_noFieldsSet() throws IOException
+  {
+    final MostFragmentedIntervalFirstPolicy policy = new MostFragmentedIntervalFirstPolicy(null, null, null, null);
+    final DefaultObjectMapper mapper = new DefaultObjectMapper();
+    final CompactionCandidateSearchPolicy policy2 =
+        mapper.readValue(mapper.writeValueAsString(policy), CompactionCandidateSearchPolicy.class);
+    Assertions.assertEquals(policy, policy2);
+  }
+
   private CompactionCandidate createCandidate(int numSegments, long avgSizeBytes)
   {
     final CompactionStatistics dummyCompactedStats = CompactionStatistics.create(1L, 1L, 1L);
@@ -201,7 +238,7 @@ private CompactionCandidate createCandidate(int numSegments, long avgSizeBytes)
         1L
     );
     return CompactionCandidate.from(List.of(SEGMENT), null)
-        .withCurrentStatus(CompactionStatus.pending(dummyCompactedStats, uncompactedStats, ""));
+                              .withCurrentStatus(CompactionStatus.pending(dummyCompactedStats, uncompactedStats, ""));
   }
 
   private void verifyCandidateIsEligible(CompactionCandidate candidate, MostFragmentedIntervalFirstPolicy policy)