Validate datasource retention rules: Reject rules that fully contain subsequent rules' interval

server/src/main/java/org/apache/druid/server/coordinator/rules/ForeverBroadcastDistributionRule.java

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.timeline.DataSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
@@ -56,6 +57,12 @@ public boolean appliesTo(Interval interval, DateTime referenceTimestamp)
     return true;
   }
 
+  @Override
+  public Interval getEligibleInterval(DateTime referenceTimestamp)
+  {
+    return Intervals.ETERNITY;
+  }
+
   @Override
   public boolean equals(Object o)
   {
@@ -75,4 +82,10 @@ public int hashCode()
   {
     return Objects.hash(getType());
   }
+
+  @Override
+  public String toString()
+  {
+    return "ForeverBroadcastDistributionRule{}";
+  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/ForeverDropRule.java

@@ -20,6 +20,7 @@
 package org.apache.druid.server.coordinator.rules;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.timeline.DataSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
@@ -46,4 +47,16 @@ public boolean appliesTo(Interval interval, DateTime referenceTimestamp)
   {
     return true;
   }
+
+  @Override
+  public Interval getEligibleInterval(DateTime referenceTimestamp)
+  {
+    return Intervals.ETERNITY;
+  }
+
+  @Override
+  public String toString()
+  {
+    return "ForeverDropRule{}";
+  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/ForeverLoadRule.java

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.timeline.DataSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
@@ -60,4 +61,18 @@ public boolean appliesTo(Interval interval, DateTime referenceTimestamp)
     return true;
   }
 
+  @Override
+  public Interval getEligibleInterval(DateTime referenceTimestamp)
+  {
+    return Intervals.ETERNITY;
+  }
+
+  @Override
+  public String toString()
+  {
+    return "ForeverLoadRule{" +
+           "tieredReplicants=" + getTieredReplicants() +
+           ", useDefaultTierForNull=" + useDefaultTierForNull() +
+           '}';
+  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/IntervalBroadcastDistributionRule.java

@@ -65,6 +65,12 @@ public Interval getInterval()
     return interval;
   }
 
+  @Override
+  public Interval getEligibleInterval(DateTime referenceTimestamp)
+  {
+    return interval;
+  }
+
   @Override
   public boolean equals(Object o)
   {
@@ -83,4 +89,12 @@ public int hashCode()
   {
     return Objects.hash(getInterval());
   }
+
+  @Override
+  public String toString()
+  {
+    return "IntervalBroadcastDistributionRule{" +
+           "interval=" + interval +
+           '}';
+  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/IntervalDropRule.java

@@ -66,6 +66,12 @@ public boolean appliesTo(Interval theInterval, DateTime referenceTimestamp)
     return interval.contains(theInterval);
   }
 
+  @Override
+  public Interval getEligibleInterval(DateTime referenceTimestamp)
+  {
+    return interval;
+  }
+
   @Override
   public boolean equals(Object o)
   {
@@ -84,4 +90,12 @@ public int hashCode()
   {
     return Objects.hash(interval);
   }
+
+  @Override
+  public String toString()
+  {
+    return "IntervalDropRule{" +
+           "interval=" + interval +
+           '}';
+  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/IntervalLoadRule.java

@@ -21,7 +21,6 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.timeline.DataSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
@@ -34,8 +33,6 @@
  */
 public class IntervalLoadRule extends LoadRule
 {
-  private static final Logger log = new Logger(IntervalLoadRule.class);
-
   private final Interval interval;
 
   @JsonCreator
@@ -74,6 +71,12 @@ public boolean appliesTo(Interval theInterval, DateTime referenceTimestamp)
     return Rules.eligibleForLoad(interval, theInterval);
   }
 
+  @Override
+  public Interval getEligibleInterval(DateTime referenceTimestamp)
+  {
+    return interval;
+  }
+
   @Override
   public boolean equals(Object o)
   {
@@ -95,4 +98,14 @@ public int hashCode()
   {
     return Objects.hash(super.hashCode(), interval);
   }
+
+  @Override
+  public String toString()
+  {
+    return "IntervalLoadRule{" +
+           "interval=" + interval +
+           ", tieredReplicants=" + getTieredReplicants() +
+           ", useDefaultTierForNull=" + useDefaultTierForNull() +
+           '}';
+  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/PeriodBroadcastDistributionRule.java

@@ -21,6 +21,8 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.JodaUtils;
 import org.apache.druid.timeline.DataSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
@@ -65,6 +67,13 @@ public boolean appliesTo(Interval interval, DateTime referenceTimestamp)
     return Rules.eligibleForLoad(period, interval, referenceTimestamp, includeFuture);
   }
 
+  @Override
+  public Interval getEligibleInterval(DateTime referenceTimestamp)
+  {
+    return includeFuture ? new Interval(referenceTimestamp.minus(period), DateTimes.utc(JodaUtils.MAX_INSTANT))
+                         : new Interval(referenceTimestamp.minus(period), referenceTimestamp);
+  }
+
   @JsonProperty
   public Period getPeriod()
   {
@@ -96,4 +105,13 @@ public int hashCode()
   {
     return Objects.hash(getPeriod(), isIncludeFuture());
   }
+
+  @Override
+  public String toString()
+  {
+    return "PeriodBroadcastDistributionRule{" +
+           "period=" + period +
+           ", includeFuture=" + includeFuture +
+           '}';
+  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/PeriodDropBeforeRule.java

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.timeline.DataSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
@@ -63,4 +64,25 @@ public boolean appliesTo(Interval theInterval, DateTime referenceTimestamp)
     final DateTime periodAgo = referenceTimestamp.minus(period);
     return theInterval.getEndMillis() <= periodAgo.getMillis();
   }
+
+  @Override
+  public Interval getEligibleInterval(DateTime referenceTimestamp)
+  {
+    final DateTime end = referenceTimestamp.minus(period);
+    if (end.isBefore(DateTimes.MIN)) {
+      // We use Long.MIN_VALUE as the start here (instead of DateTimes.MIN) when end is < DateTimes.MIN because the
+      // resulting interval will be invalid where start > end. This is true for referenceTimestamp = DateTimes.MIN.
+      return new Interval(DateTimes.utc(Long.MIN_VALUE), end);
+    } else {
+      return new Interval(DateTimes.MIN, end);
+    }
+  }
+
+  @Override
+  public String toString()
+  {
+    return "PeriodDropBeforeRule{" +
+           "period=" + period +
+           '}';
+  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/PeriodDropRule.java

@@ -21,12 +21,14 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.timeline.DataSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 import org.joda.time.Period;
 
 /**
+ *
  */
 public class PeriodDropRule extends DropRule
 {
@@ -80,4 +82,20 @@ public boolean appliesTo(Interval theInterval, DateTime referenceTimestamp)
       return currInterval.contains(theInterval);
     }
   }
+
+  @Override
+  public Interval getEligibleInterval(DateTime referenceTimestamp)
+  {
+    return includeFuture ? new Interval(referenceTimestamp.minus(period), DateTimes.MAX)
+                         : new Interval(referenceTimestamp.minus(period), referenceTimestamp);
+  }
+
+  @Override
+  public String toString()
+  {
+    return "PeriodDropRule{" +
+           "period=" + period +
+           ", includeFuture=" + includeFuture +
+           '}';
+  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/PeriodLoadRule.java

@@ -21,7 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.timeline.DataSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
@@ -35,7 +35,6 @@
  */
 public class PeriodLoadRule extends LoadRule
 {
-  private static final Logger log = new Logger(PeriodLoadRule.class);
   static final boolean DEFAULT_INCLUDE_FUTURE = true;
 
   private final Period period;
@@ -85,6 +84,13 @@ public boolean appliesTo(Interval interval, DateTime referenceTimestamp)
     return Rules.eligibleForLoad(period, interval, referenceTimestamp, includeFuture);
   }
 
+  @Override
+  public Interval getEligibleInterval(DateTime referenceTimestamp)
+  {
+    return includeFuture ? new Interval(referenceTimestamp.minus(period), DateTimes.MAX)
+                         : new Interval(referenceTimestamp.minus(period), referenceTimestamp);
+  }
+
   @Override
   public boolean equals(Object o)
   {
@@ -106,4 +112,15 @@ public int hashCode()
   {
     return Objects.hash(super.hashCode(), period, includeFuture);
   }
+
+  @Override
+  public String toString()
+  {
+    return "PeriodLoadRule{" +
+           "period=" + period +
+           ", includeFuture=" + includeFuture +
+           ", tieredReplicants=" + getTieredReplicants() +
+           ", useDefaultTierForNull=" + useDefaultTierForNull() +
+           '}';
+  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/Rule.java

@@ -50,4 +50,11 @@ public interface Rule
   boolean appliesTo(Interval interval, DateTime referenceTimestamp);
 
   void run(DataSegment segment, SegmentActionHandler segmentHandler);
+
+  /**
+   * Returns the interval eligible for this rule. The interval must be computed based on the rule type
+   * optionally using {@code referenceTimestamp}. {@code referenceTimestamp} must be a {@link DateTime}
+   * between [{@link org.apache.druid.java.util.common.DateTimes#MIN}, {@link org.apache.druid.java.util.common.DateTimes#MAX}].
+   */
+  Interval getEligibleInterval(DateTime referenceTimestamp);
 }