Fix filterSegments for TimeBoundary and DataSourceMetadata queries

core/src/main/java/org/apache/druid/timeline/LogicalSegment.java

@@ -22,8 +22,26 @@
 import org.apache.druid.guice.annotations.PublicApi;
 import org.joda.time.Interval;
 
+/**
+ * A logical segment can represent an entire segment or a part of a segment. As a result, it can have a different
+ * interval from its actual base segment. {@link #getInterval()} and {@link #getTrueInterval()} return the interval of
+ * this logical segment and the interval of the base segment, respectively.
+ *
+ * For example, suppose we have 2 segments as below:
+ *
+ * - Segment A has an interval of 2017/2018.
+ * - Segment B has an interval of 2017-08-01/2017-08-02.
+ *
+ * For these segments, {@link VersionedIntervalTimeline#lookup} returns 3 segments as below:
+ *
+ * - interval of 2017/2017-08-01 (trueInterval: 2017/2018)
+ * - interval of 2017-08-01/2017-08-02 (trueInterval: 2017-08-01/2017-08-02)
+ * - interval of 2017-08-02/2018 (trueInterval: 2017/2018)
+ */
 @PublicApi
 public interface LogicalSegment
 {
   Interval getInterval();
+
+  Interval getTrueInterval();
 }

core/src/main/java/org/apache/druid/timeline/TimelineObjectHolder.java

@@ -19,6 +19,7 @@
 
 package org.apache.druid.timeline;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.druid.timeline.partition.PartitionHolder;
 import org.joda.time.Interval;
 
@@ -27,16 +28,25 @@
 public class TimelineObjectHolder<VersionType, ObjectType> implements LogicalSegment
 {
   private final Interval interval;
+  private final Interval trueInterval;
   private final VersionType version;
   private final PartitionHolder<ObjectType> object;
 
+  @VisibleForTesting
+  public TimelineObjectHolder(Interval interval, VersionType version, PartitionHolder<ObjectType> object)
+  {
+    this(interval, interval, version, object);
+  }
+
   public TimelineObjectHolder(
       Interval interval,
+      Interval trueInterval,
       VersionType version,
       PartitionHolder<ObjectType> object
   )
   {
     this.interval = interval;
+    this.trueInterval = trueInterval;
     this.version = version;
     this.object = object;
   }
@@ -47,6 +57,12 @@ public Interval getInterval()
     return interval;
   }
 
+  @Override
+  public Interval getTrueInterval()
+  {
+    return trueInterval;
+  }
+
   public VersionType getVersion()
   {
     return version;
@@ -62,6 +78,7 @@ public String toString()
   {
     return "TimelineObjectHolder{" +
            "interval=" + interval +
+           ", trueInterval=" + trueInterval +
            ", version=" + version +
            ", object=" + object +
            '}';

core/src/main/java/org/apache/druid/timeline/VersionedIntervalTimeline.java

@@ -300,6 +300,7 @@ public TimelineObjectHolder<VersionType, ObjectType> last()
   private TimelineObjectHolder<VersionType, ObjectType> timelineEntryToObjectHolder(TimelineEntry entry)
   {
     return new TimelineObjectHolder<>(
+        entry.getTrueInterval(),
         entry.getTrueInterval(),
         entry.getVersion(),
         new PartitionHolder<>(entry.getPartitionHolder())
@@ -586,10 +587,11 @@ private List<TimelineObjectHolder<VersionType, ObjectType>> lookup(Interval inte
 
       if (timelineInterval.overlaps(interval)) {
         retVal.add(
-            new TimelineObjectHolder<VersionType, ObjectType>(
+            new TimelineObjectHolder<>(
                 timelineInterval,
+                val.getTrueInterval(),
                 val.getVersion(),
-                new PartitionHolder<ObjectType>(val.getPartitionHolder())
+                new PartitionHolder<>(val.getPartitionHolder())
             )
         );
       }
@@ -604,8 +606,9 @@ private List<TimelineObjectHolder<VersionType, ObjectType>> lookup(Interval inte
                                                                .isAfter(firstEntry.getInterval().getStart())) {
       retVal.set(
           0,
-          new TimelineObjectHolder<VersionType, ObjectType>(
+          new TimelineObjectHolder<>(
               new Interval(interval.getStart(), firstEntry.getInterval().getEnd()),
+              firstEntry.getTrueInterval(),
               firstEntry.getVersion(),
               firstEntry.getObject()
           )
@@ -616,8 +619,9 @@ private List<TimelineObjectHolder<VersionType, ObjectType>> lookup(Interval inte
     if (interval.overlaps(lastEntry.getInterval()) && interval.getEnd().isBefore(lastEntry.getInterval().getEnd())) {
       retVal.set(
           retVal.size() - 1,
-          new TimelineObjectHolder<VersionType, ObjectType>(
+          new TimelineObjectHolder<>(
               new Interval(lastEntry.getInterval().getStart(), interval.getEnd()),
+              lastEntry.getTrueInterval(),
               lastEntry.getVersion(),
               lastEntry.getObject()
           )

processing/src/main/java/org/apache/druid/query/datasourcemetadata/DataSourceQueryQueryToolChest.java

@@ -22,9 +22,6 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.google.common.base.Function;
 import com.google.common.base.Functions;
-import com.google.common.base.Predicate;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
 import com.google.inject.Inject;
 import org.apache.druid.java.util.common.guava.Sequence;
 import org.apache.druid.java.util.common.guava.Sequences;
@@ -42,6 +39,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  */
@@ -68,19 +66,9 @@ public <T extends LogicalSegment> List<T> filterSegments(DataSourceMetadataQuery
 
     final T max = segments.get(segments.size() - 1);
 
-    return Lists.newArrayList(
-        Iterables.filter(
-            segments,
-            new Predicate<T>()
-            {
-              @Override
-              public boolean apply(T input)
-              {
-                return max != null && input.getInterval().overlaps(max.getInterval());
-              }
-            }
-        )
-    );
+    return segments.stream()
+                   .filter(input -> max != null && input.getInterval().overlaps(max.getTrueInterval()))
+                   .collect(Collectors.toList());
   }
 
   @Override

processing/src/main/java/org/apache/druid/query/timeboundary/TimeBoundaryQueryQueryToolChest.java

@@ -23,8 +23,6 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Function;
 import com.google.common.base.Functions;
-import com.google.common.base.Predicate;
-import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import com.google.inject.Inject;
 import org.apache.druid.java.util.common.DateTimes;
@@ -46,6 +44,7 @@
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  */
@@ -85,20 +84,10 @@ public <T extends LogicalSegment> List<T> filterSegments(TimeBoundaryQuery query
     final T min = query.isMaxTime() ? null : segments.get(0);
     final T max = query.isMinTime() ? null : segments.get(segments.size() - 1);
 
-    return Lists.newArrayList(
-        Iterables.filter(
-            segments,
-            new Predicate<T>()
-            {
-              @Override
-              public boolean apply(T input)
-              {
-                return (min != null && input.getInterval().overlaps(min.getInterval())) ||
-                       (max != null && input.getInterval().overlaps(max.getInterval()));
-              }
-            }
-        )
-    );
+    return segments.stream()
+                   .filter(input -> (min != null && input.getInterval().overlaps(min.getTrueInterval())) ||
+                                    (max != null && input.getInterval().overlaps(max.getTrueInterval())))
+                   .collect(Collectors.toList());
   }
 
   @Override