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

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

@@ -56,6 +56,12 @@ public boolean appliesTo(Interval interval, DateTime referenceTimestamp)
     return true;
   }
 
+  @Override
+  public Interval getInterval(DateTime referenceTimestamp)
+  {
+    return Rules.FOREVER_INTERVAL;
+  }
+
   @Override
   public boolean equals(Object o)
   {
@@ -75,4 +81,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

@@ -46,4 +46,16 @@ public boolean appliesTo(Interval interval, DateTime referenceTimestamp)
   {
     return true;
   }
+
+  @Override
+  public Interval getInterval(DateTime referenceTimestamp)
+  {
+    return Rules.FOREVER_INTERVAL;
+  }
+
+  @Override
+  public String toString()
+  {
+    return "ForeverDropRule{}";
+  }
 }

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

@@ -60,4 +60,18 @@ public boolean appliesTo(Interval interval, DateTime referenceTimestamp)
     return true;
   }
 
+  @Override
+  public Interval getInterval(DateTime referenceTimestamp)
+  {
+    return Rules.FOREVER_INTERVAL;
+  }
+
+  @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 getInterval(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 getInterval(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 getInterval(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

@@ -65,6 +65,12 @@ public boolean appliesTo(Interval interval, DateTime referenceTimestamp)
     return Rules.eligibleForLoad(period, interval, referenceTimestamp, includeFuture);
   }
 
+  @Override
+  public Interval getInterval(DateTime referenceTimestamp)
+  {
+    return new Interval(referenceTimestamp.minus(period), referenceTimestamp);
+  }
+
   @JsonProperty
   public Period getPeriod()
   {
@@ -96,4 +102,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,18 @@ public boolean appliesTo(Interval theInterval, DateTime referenceTimestamp)
     final DateTime periodAgo = referenceTimestamp.minus(period);
     return theInterval.getEndMillis() <= periodAgo.getMillis();
   }
+
+  @Override
+  public Interval getInterval(DateTime referenceTimestamp)
+  {
+    return new Interval(DateTimes.utc(Long.MIN_VALUE), referenceTimestamp.minus(period));
+  }
+
+  @Override
+  public String toString()
+  {
+    return "PeriodDropBeforeRule{" +
+           "period=" + period +
+           '}';
+  }
 }

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

@@ -80,4 +80,19 @@ public boolean appliesTo(Interval theInterval, DateTime referenceTimestamp)
       return currInterval.contains(theInterval);
     }
   }
+
+  @Override
+  public Interval getInterval(DateTime referenceTimestamp)
+  {
+    return 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,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;
@@ -35,7 +34,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 +83,12 @@ public boolean appliesTo(Interval interval, DateTime referenceTimestamp)
     return Rules.eligibleForLoad(period, interval, referenceTimestamp, includeFuture);
   }
 
+  @Override
+  public Interval getInterval(DateTime referenceTimestamp)
+  {
+    return new Interval(referenceTimestamp.minus(period), referenceTimestamp);
+  }
+
   @Override
   public boolean equals(Object o)
   {
@@ -106,4 +110,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,6 @@ public interface Rule
   boolean appliesTo(Interval interval, DateTime referenceTimestamp);
 
   void run(DataSegment segment, SegmentActionHandler segmentHandler);
+
+  Interval getInterval(DateTime referenceTimestamp);
 }

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

@@ -19,12 +19,15 @@
 
 package org.apache.druid.server.coordinator.rules;
 
+import org.apache.druid.java.util.common.DateTimes;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 import org.joda.time.Period;
 
 public class Rules
 {
+  public static final Interval FOREVER_INTERVAL = new Interval(DateTimes.utc(Long.MIN_VALUE), DateTimes.utc(Long.MAX_VALUE));
+
   public static boolean eligibleForLoad(Interval src, Interval target)
   {
     return src.overlaps(target);

server/src/main/java/org/apache/druid/server/http/RulesResource.java

@@ -25,11 +25,15 @@
 import org.apache.druid.audit.AuditEntry;
 import org.apache.druid.audit.AuditInfo;
 import org.apache.druid.audit.AuditManager;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.metadata.MetadataRuleManager;
 import org.apache.druid.server.coordinator.rules.Rule;
+import org.apache.druid.server.coordinator.rules.Rules;
 import org.apache.druid.server.http.security.RulesResourceFilter;
 import org.apache.druid.server.http.security.StateResourceFilter;
+import org.joda.time.DateTime;
 import org.joda.time.Interval;
 
 import javax.servlet.http.HttpServletRequest;
@@ -108,6 +112,7 @@ public Response setDatasourceRules(
   )
   {
     try {
+      validateRules(rules);
       final AuditInfo auditInfo = new AuditInfo(author, comment, req.getRemoteAddr());
       if (databaseRuleManager.overrideRule(dataSourceName, rules, auditInfo)) {
         return Response.ok().build();
@@ -181,4 +186,37 @@ private List<AuditEntry> getRuleHistory(
     return auditManager.fetchAuditHistory(AUDIT_HISTORY_TYPE, theInterval);
   }
 
+  /**
+   * Validate rules. Throws an exception if a rule contain an interval that will overshadow another rules' interval.
+   * Rules that will be evaluated at some point are considered to be non-overshadowing.
+   * @param rules Datasource rules.
+   */
+  private void validateRules(final List<Rule> rules)
+  {
+    if (rules == null) {
+      return;
+    }
+    final DateTime now = DateTimes.nowUtc();
+    for (int i = 0; i < rules.size() - 1; i++) {
+      final Rule currRule = rules.get(i);
+      final Rule nextRule = rules.get(i + 1);
+      final Interval currInterval = currRule.getInterval(now);
+      final Interval nextInterval = nextRule.getInterval(now);
+      if (currInterval.contains(nextInterval)) {
+        // If the current rule overshaows the next rule even at the intervals' boundaries, then we know that the next
+        // rule will always be a no-op. Also, a forever rule spans eternity and overshadows everything that follows it.
+        if (Rules.FOREVER_INTERVAL.equals(currInterval) ||
+                    (currRule.getInterval(currInterval.getStart()).contains(nextRule.getInterval(currInterval.getStart()))
+                     && currRule.getInterval(currInterval.getEnd()).contains(nextRule.getInterval(currInterval.getEnd())))) {
+          throw InvalidInput.exception(
+              "Rule[%s] has an interval that contains interval for rule[%s]. The interval[%s] also covers interval[%s].",
+              currRule,
+              nextRule,
+              currInterval,
+              nextInterval
+          );
+        }
+      }
+    }
+  }
 }