Reconcile terminology and method naming to 'used/unused segments'; Rename MetadataSegmentManager to MetadataSegmentsManager

benchmarks/src/main/java/org/apache/druid/server/coordinator/NewestSegmentFirstPolicyBenchmark.java

@@ -23,9 +23,9 @@
 import com.google.common.collect.ImmutableMap;
 import org.apache.druid.client.DataSourcesSnapshot;
 import org.apache.druid.java.util.common.DateTimes;
-import org.apache.druid.server.coordinator.helper.CompactionSegmentIterator;
-import org.apache.druid.server.coordinator.helper.CompactionSegmentSearchPolicy;
-import org.apache.druid.server.coordinator.helper.NewestSegmentFirstPolicy;
+import org.apache.druid.server.coordinator.duty.CompactionSegmentIterator;
+import org.apache.druid.server.coordinator.duty.CompactionSegmentSearchPolicy;
+import org.apache.druid.server.coordinator.duty.NewestSegmentFirstPolicy;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.timeline.VersionedIntervalTimeline;
 import org.apache.druid.timeline.partition.NumberedShardSpec;

core/src/main/java/org/apache/druid/java/util/metrics/AllocationMetricCollectors.java

@@ -22,7 +22,6 @@
 import org.apache.druid.java.util.common.logger.Logger;
 
 import javax.annotation.Nullable;
-
 import java.lang.management.ManagementFactory;
 import java.lang.management.ThreadMXBean;
 import java.lang.reflect.Method;

core/src/main/java/org/apache/druid/metadata/MetadataStorageConnector.java

@@ -19,6 +19,7 @@
 
 package org.apache.druid.metadata;
 
+import javax.annotation.Nullable;
 import java.util.List;
 
 /**
@@ -36,7 +37,7 @@ Void insertOrUpdate(
       byte[] value
   );
 
-  byte[] lookup(
+  @Nullable byte[] lookup(
       String tableName,
       String keyColumn,
       String valueColumn,

core/src/main/java/org/apache/druid/metadata/MetadataStorageConnectorConfig.java

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.druid.java.util.common.StringUtils;
+
 import java.util.Properties;
 
 /**

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

@@ -37,7 +37,6 @@
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.IdentityHashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -49,28 +48,44 @@
 import java.util.TreeMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
 /**
  * VersionedIntervalTimeline is a data structure that manages objects on a specific timeline.
  *
- * It associates a jodatime Interval and a generically-typed version with the object that is being stored.
+ * It associates an {@link Interval} and a generically-typed version with the object that is being stored.
  *
  * In the event of overlapping timeline entries, timeline intervals may be chunked. The underlying data associated
  * with a timeline entry remains unchanged when chunking occurs.
  *
- * After loading objects via the add() method, the lookup(Interval) method can be used to get the list of the most
- * recent objects (according to the version) that match the given interval.  The intent is that objects represent
- * a certain time period and when you do a lookup(), you are asking for all of the objects that you need to look
- * at in order to get a correct answer about that time period.
+ * After loading objects via the {@link #add} method, the {@link #lookup(Interval)} method can be used to get the list
+ * of the most recent objects (according to the version) that match the given interval. The intent is that objects
+ * represent a certain time period and when you do a {@link #lookup(Interval)}, you are asking for all of the objects
+ * that you need to look at in order to get a correct answer about that time period.
  *
- * The findFullyOvershadowed() method returns a list of objects that will never be returned by a call to lookup() because
- * they are overshadowed by some other object.  This can be used in conjunction with the add() and remove() methods
- * to achieve "atomic" updates.  First add new items, then check if those items caused anything to be overshadowed, if
- * so, remove the overshadowed elements and you have effectively updated your data set without any user impact.
+ * The {@link #findFullyOvershadowed} method returns a list of objects that will never be returned by a call to {@link
+ * #lookup} because they are overshadowed by some other object. This can be used in conjunction with the {@link #add}
+ * and {@link #remove} methods to achieve "atomic" updates. First add new items, then check if those items caused
+ * anything to be overshadowed, if so, remove the overshadowed elements and you have effectively updated your data set
+ * without any user impact.
  */
-public class VersionedIntervalTimeline<VersionType, ObjectType extends Overshadowable<ObjectType>> implements TimelineLookup<VersionType, ObjectType>
+public class VersionedIntervalTimeline<VersionType, ObjectType extends Overshadowable<ObjectType>>
+    implements TimelineLookup<VersionType, ObjectType>
 {
+  public static VersionedIntervalTimeline<String, DataSegment> forSegments(Iterable<DataSegment> segments)
+  {
+    return forSegments(segments.iterator());
+  }
+
+  public static VersionedIntervalTimeline<String, DataSegment> forSegments(Iterator<DataSegment> segments)
+  {
+    final VersionedIntervalTimeline<String, DataSegment> timeline =
+        new VersionedIntervalTimeline<>(Comparator.naturalOrder());
+    addSegments(timeline, segments);
+    return timeline;
+  }
+
   private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
 
   // Below timelines stores only *visible* timelineEntries
@@ -95,19 +110,6 @@ public VersionedIntervalTimeline(Comparator<? super VersionType> versionComparat
     this.versionComparator = versionComparator;
   }
 
-  public static VersionedIntervalTimeline<String, DataSegment> forSegments(Iterable<DataSegment> segments)
-  {
-    return forSegments(segments.iterator());
-  }
-
-  public static VersionedIntervalTimeline<String, DataSegment> forSegments(Iterator<DataSegment> segments)
-  {
-    final VersionedIntervalTimeline<String, DataSegment> timeline =
-        new VersionedIntervalTimeline<>(Comparator.naturalOrder());
-    addSegments(timeline, segments);
-    return timeline;
-  }
-
   public static void addSegments(
       VersionedIntervalTimeline<String, DataSegment> timeline,
       Iterator<DataSegment> segments
@@ -147,6 +149,11 @@ public Collection<ObjectType> iterateAllObjects()
     );
   }
 
+  public int getNumObjects()
+  {
+    return numObjects.get();
+  }
+
   public void add(final Interval interval, VersionType version, PartitionChunk<ObjectType> object)
   {
     addAll(Iterators.singletonIterator(object), o -> interval, o -> version);
@@ -354,62 +361,69 @@ public Set<TimelineObjectHolder<VersionType, ObjectType>> findFullyOvershadowed(
     lock.readLock().lock();
     try {
       // 1. Put all timelineEntries and remove all visible entries to find out only non-visible timelineEntries.
-      final Map<Interval, Map<VersionType, TimelineEntry>> overShadowed = new HashMap<>();
-      for (Map.Entry<Interval, TreeMap<VersionType, TimelineEntry>> versionEntry : allTimelineEntries.entrySet()) {
-        @SuppressWarnings("unchecked")
-        Map<VersionType, TimelineEntry> versionCopy = (TreeMap) versionEntry.getValue().clone();
-        overShadowed.put(versionEntry.getKey(), versionCopy);
-      }
-
-      for (Entry<Interval, TimelineEntry> entry : completePartitionsTimeline.entrySet()) {
-        Map<VersionType, TimelineEntry> versionEntry = overShadowed.get(entry.getValue().getTrueInterval());
-        if (versionEntry != null) {
-          versionEntry.remove(entry.getValue().getVersion());
-          if (versionEntry.isEmpty()) {
-            overShadowed.remove(entry.getValue().getTrueInterval());
-          }
-        }
-      }
-
-      for (Entry<Interval, TimelineEntry> entry : incompletePartitionsTimeline.entrySet()) {
-        Map<VersionType, TimelineEntry> versionEntry = overShadowed.get(entry.getValue().getTrueInterval());
-        if (versionEntry != null) {
-          versionEntry.remove(entry.getValue().getVersion());
-          if (versionEntry.isEmpty()) {
-            overShadowed.remove(entry.getValue().getTrueInterval());
-          }
-        }
-      }
-
-      final Set<TimelineObjectHolder<VersionType, ObjectType>> retVal = new HashSet<>();
-      for (Entry<Interval, Map<VersionType, TimelineEntry>> versionEntry : overShadowed.entrySet()) {
-        for (Entry<VersionType, TimelineEntry> entry : versionEntry.getValue().entrySet()) {
-          final TimelineEntry timelineEntry = entry.getValue();
-          retVal.add(timelineEntryToObjectHolder(timelineEntry));
-        }
-      }
+      final Map<Interval, Map<VersionType, TimelineEntry>> overshadowedPartitionsTimeline =
+          computeOvershadowedPartitionsTimeline();
+
+      final Set<TimelineObjectHolder<VersionType, ObjectType>> overshadowedObjects = overshadowedPartitionsTimeline
+          .values()
+          .stream()
+          .flatMap(
+              (Map<VersionType, TimelineEntry> entry) -> entry.values().stream().map(this::timelineEntryToObjectHolder)
+          )
+          .collect(Collectors.toSet());
 
-      // 2. Visible timelineEntries can also have overshadowed segments. Add them to the result too.
+      // 2. Visible timelineEntries can also have overshadowed objects. Add them to the result too.
       for (TimelineEntry entry : incompletePartitionsTimeline.values()) {
-        final List<PartitionChunk<ObjectType>> entryOvershadowed = entry.partitionHolder.getOvershadowed();
-        if (!entryOvershadowed.isEmpty()) {
-          retVal.add(
+        final List<PartitionChunk<ObjectType>> overshadowedEntries = entry.partitionHolder.getOvershadowed();
+        if (!overshadowedEntries.isEmpty()) {
+          overshadowedObjects.add(
               new TimelineObjectHolder<>(
                   entry.trueInterval,
                   entry.version,
-                  new PartitionHolder<>(entryOvershadowed)
+                  new PartitionHolder<>(overshadowedEntries)
               )
           );
         }
       }
 
-      return retVal;
+      return overshadowedObjects;
     }
     finally {
       lock.readLock().unlock();
     }
   }
 
+  private Map<Interval, Map<VersionType, TimelineEntry>> computeOvershadowedPartitionsTimeline()
+  {
+    final Map<Interval, Map<VersionType, TimelineEntry>> overshadowedPartitionsTimeline = new HashMap<>();
+    allTimelineEntries.forEach((Interval interval, TreeMap<VersionType, TimelineEntry> versionEntry) -> {
+      @SuppressWarnings("unchecked")
+      Map<VersionType, TimelineEntry> versionEntryCopy = (TreeMap) versionEntry.clone();
+      overshadowedPartitionsTimeline.put(interval, versionEntryCopy);
+    });
+
+    for (TimelineEntry entry : completePartitionsTimeline.values()) {
+      overshadowedPartitionsTimeline.computeIfPresent(
+          entry.getTrueInterval(),
+          (Interval interval, Map<VersionType, TimelineEntry> versionEntry) -> {
+            versionEntry.remove(entry.getVersion());
+            return versionEntry.isEmpty() ? null : versionEntry;
+          }
+      );
+    }
+
+    for (TimelineEntry entry : incompletePartitionsTimeline.values()) {
+      overshadowedPartitionsTimeline.computeIfPresent(
+          entry.getTrueInterval(),
+          (Interval interval, Map<VersionType, TimelineEntry> versionEntry) -> {
+            versionEntry.remove(entry.getVersion());
+            return versionEntry.isEmpty() ? null : versionEntry;
+          }
+      );
+    }
+    return overshadowedPartitionsTimeline;
+  }
+
   public boolean isOvershadowed(Interval interval, VersionType version, ObjectType object)
   {
     lock.readLock().lock();

core/src/main/java/org/apache/druid/timeline/partition/OvershadowableManager.java

@@ -88,22 +88,22 @@ enum State
 
   // (start partitionId, end partitionId) -> minorVersion -> atomicUpdateGroup
   private final TreeMap<RootPartitionRange, Short2ObjectSortedMap<AtomicUpdateGroup<T>>> standbyGroups;
-  private final TreeMap<RootPartitionRange, Short2ObjectSortedMap<AtomicUpdateGroup<T>>> visibleGroup;
+  private final TreeMap<RootPartitionRange, Short2ObjectSortedMap<AtomicUpdateGroup<T>>> visibleGroups;
   private final TreeMap<RootPartitionRange, Short2ObjectSortedMap<AtomicUpdateGroup<T>>> overshadowedGroups;
 
   OvershadowableManager()
   {
     this.knownPartitionChunks = new HashMap<>();
     this.standbyGroups = new TreeMap<>();
-    this.visibleGroup = new TreeMap<>();
+    this.visibleGroups = new TreeMap<>();
     this.overshadowedGroups = new TreeMap<>();
   }
 
   OvershadowableManager(OvershadowableManager<T> other)
   {
     this.knownPartitionChunks = new HashMap<>(other.knownPartitionChunks);
     this.standbyGroups = new TreeMap<>(other.standbyGroups);
-    this.visibleGroup = new TreeMap<>(other.visibleGroup);
+    this.visibleGroups = new TreeMap<>(other.visibleGroups);
     this.overshadowedGroups = new TreeMap<>(other.overshadowedGroups);
   }
 
@@ -123,7 +123,7 @@ private TreeMap<RootPartitionRange, Short2ObjectSortedMap<AtomicUpdateGroup<T>>>
       case STANDBY:
         return standbyGroups;
       case VISIBLE:
-        return visibleGroup;
+        return visibleGroups;
       case OVERSHADOWED:
         return overshadowedGroups;
       default:
@@ -669,7 +669,7 @@ boolean addChunk(PartitionChunk<T> chunk)
           final AtomicUpdateGroup<T> newAtomicUpdateGroup = new AtomicUpdateGroup<>(chunk);
 
           // Decide the initial state of the new atomicUpdateGroup
-          final boolean overshadowed = visibleGroup
+          final boolean overshadowed = visibleGroups
               .values()
               .stream()
               .flatMap(map -> map.values().stream())
@@ -785,7 +785,7 @@ private List<AtomicUpdateGroup<T>> findLatestNonFullyAvailableAtomicUpdateGroups
     }
 
     final List<AtomicUpdateGroup<T>> visibles = new ArrayList<>();
-    for (Short2ObjectSortedMap<AtomicUpdateGroup<T>> map : manager.visibleGroup.values()) {
+    for (Short2ObjectSortedMap<AtomicUpdateGroup<T>> map : manager.visibleGroups.values()) {
       visibles.addAll(map.values());
     }
     return visibles;
@@ -807,7 +807,7 @@ private List<AtomicUpdateGroup<T>> findLatestFullyAvailableOvershadowedAtomicUpd
 
     final OvershadowableManager<T> manager = new OvershadowableManager<>(overshadowedGroups);
     final List<AtomicUpdateGroup<T>> visibles = new ArrayList<>();
-    for (Short2ObjectSortedMap<AtomicUpdateGroup<T>> map : manager.visibleGroup.values()) {
+    for (Short2ObjectSortedMap<AtomicUpdateGroup<T>> map : manager.visibleGroups.values()) {
       for (AtomicUpdateGroup<T> atomicUpdateGroup : map.values()) {
         if (!atomicUpdateGroup.isFull()) {
           return Collections.emptyList();
@@ -885,12 +885,12 @@ PartitionChunk<T> removeChunk(PartitionChunk<T> partitionChunk)
 
   public boolean isEmpty()
   {
-    return visibleGroup.isEmpty();
+    return visibleGroups.isEmpty();
   }
 
   public boolean isComplete()
   {
-    return visibleGroup.values().stream().allMatch(map -> Iterables.getOnlyElement(map.values()).isFull());
+    return visibleGroups.values().stream().allMatch(map -> Iterables.getOnlyElement(map.values()).isFull());
   }
 
   @Nullable
@@ -915,7 +915,7 @@ PartitionChunk<T> getChunk(int partitionId)
 
   List<PartitionChunk<T>> getVisibleChunks()
   {
-    return getAllChunks(visibleGroup);
+    return getAllChunks(visibleGroups);
   }
 
   List<PartitionChunk<T>> getOvershadowedChunks()
@@ -954,14 +954,14 @@ public boolean equals(Object o)
     OvershadowableManager<?> that = (OvershadowableManager<?>) o;
     return Objects.equals(knownPartitionChunks, that.knownPartitionChunks) &&
            Objects.equals(standbyGroups, that.standbyGroups) &&
-           Objects.equals(visibleGroup, that.visibleGroup) &&
+           Objects.equals(visibleGroups, that.visibleGroups) &&
            Objects.equals(overshadowedGroups, that.overshadowedGroups);
   }
 
   @Override
   public int hashCode()
   {
-    return Objects.hash(knownPartitionChunks, standbyGroups, visibleGroup, overshadowedGroups);
+    return Objects.hash(knownPartitionChunks, standbyGroups, visibleGroups, overshadowedGroups);
   }
 
   @Override
@@ -970,7 +970,7 @@ public String toString()
     return "OvershadowableManager{" +
            "knownPartitionChunks=" + knownPartitionChunks +
            ", standbyGroups=" + standbyGroups +
-           ", visibleGroup=" + visibleGroup +
+           ", visibleGroups=" + visibleGroups +
            ", overshadowedGroups=" + overshadowedGroups +
            '}';
   }