From 7e67397b5a35f670e41f1a88b910f13251535077 Mon Sep 17 00:00:00 2001
From: Himanshu <g.himanshu@gmail.com>
Date: Wed, 25 May 2016 13:01:22 -0500
Subject: [PATCH] fix-3010: look through all versions to find the set with
 complete partitions (#3013)

---
 .../timeline/VersionedIntervalTimeline.java   | 13 +++-
 .../VersionedIntervalTimelineTest.java        | 63 ++++++++++++++++++-
 2 files changed, 72 insertions(+), 4 deletions(-)

diff --git a/common/src/main/java/io/druid/timeline/VersionedIntervalTimeline.java b/common/src/main/java/io/druid/timeline/VersionedIntervalTimeline.java
index 19219f59d16..04918690797 100644
--- a/common/src/main/java/io/druid/timeline/VersionedIntervalTimeline.java
+++ b/common/src/main/java/io/druid/timeline/VersionedIntervalTimeline.java
@@ -422,9 +422,16 @@ public class VersionedIntervalTimeline<VersionType, ObjectType> implements Timel
 
     for (Map.Entry<Interval, TreeMap<VersionType, TimelineEntry>> versionEntry : allTimelineEntries.entrySet()) {
       if (versionEntry.getKey().overlap(interval) != null) {
-        TimelineEntry timelineEntry = versionEntry.getValue().lastEntry().getValue();
-        if (timelineEntry.getPartitionHolder().isComplete() || incompleteOk) {
-          add(timeline, versionEntry.getKey(), timelineEntry);
+        if (incompleteOk) {
+          add(timeline, versionEntry.getKey(), versionEntry.getValue().lastEntry().getValue());
+        } else {
+          for (VersionType ver : versionEntry.getValue().descendingKeySet()) {
+            TimelineEntry timelineEntry = versionEntry.getValue().get(ver);
+            if (timelineEntry.getPartitionHolder().isComplete()) {
+              add(timeline, versionEntry.getKey(), timelineEntry);
+              break;
+            }
+          }
         }
       }
     }
diff --git a/common/src/test/java/io/druid/timeline/VersionedIntervalTimelineTest.java b/common/src/test/java/io/druid/timeline/VersionedIntervalTimelineTest.java
index bc88e39d905..de0bb98a9fd 100644
--- a/common/src/test/java/io/druid/timeline/VersionedIntervalTimelineTest.java
+++ b/common/src/test/java/io/druid/timeline/VersionedIntervalTimelineTest.java
@@ -1504,7 +1504,68 @@ public class VersionedIntervalTimelineTest
 
     Assert.assertTrue(timeline.lookup(Interval.parse("1970/1980")).isEmpty());
   }
-  
+
+  // https://github.com/druid-io/druid/issues/3010
+  @Test
+  public void testRemoveIncompleteKeepsComplete() throws Exception
+  {
+    timeline = makeStringIntegerTimeline();
+
+    add("2011-04-01/2011-04-02", "1", IntegerPartitionChunk.make(null, 1, 0, 77));
+    add("2011-04-01/2011-04-02", "1", IntegerPartitionChunk.make(1, null, 1, 88));
+    add("2011-04-01/2011-04-02", "2", IntegerPartitionChunk.make(null, 1, 0, 99));
+
+    assertValues(
+        ImmutableList.of(
+            createExpected("2011-04-01/2011-04-02", "1",
+                           Arrays.<PartitionChunk<Integer>>asList(
+                               IntegerPartitionChunk.make(null, 1, 0, 77),
+                               IntegerPartitionChunk.make(1, null, 1, 88)
+                           )
+            )
+        ),
+        timeline.lookup(new Interval("2011-04-01/2011-04-02"))
+    );
+
+    add("2011-04-01/2011-04-02", "3", IntegerPartitionChunk.make(null, 1, 0, 110));
+
+    assertValues(
+        ImmutableList.of(
+            createExpected("2011-04-01/2011-04-02", "1",
+                           Arrays.<PartitionChunk<Integer>>asList(
+                               IntegerPartitionChunk.make(null, 1, 0, 77),
+                               IntegerPartitionChunk.make(1, null, 1, 88)
+                           )
+            )
+        ),
+        timeline.lookup(new Interval("2011-04-01/2011-04-02"))
+    );
+    assertValues(
+        Sets.newHashSet(
+            createExpected("2011-04-01/2011-04-02", "2",
+                           Arrays.<PartitionChunk<Integer>>asList(
+                               IntegerPartitionChunk.make(null, 1, 0, 99)
+                           )
+            )
+        ),
+        timeline.findOvershadowed()
+    );
+
+    testRemove();
+
+    assertValues(
+        ImmutableList.of(
+            createExpected("2011-04-01/2011-04-02", "1",
+                           Arrays.<PartitionChunk<Integer>>asList(
+                               IntegerPartitionChunk.make(null, 1, 0, 77),
+                               IntegerPartitionChunk.make(1, null, 1, 88)
+                           )
+            )
+        ),
+        timeline.lookup(new Interval("2011-04-01/2011-04-02"))
+    );
+  }
+
   private Pair<Interval, Pair<String, PartitionHolder<Integer>>> createExpected(
       String intervalString,
       String version,