Add is_overshadowed column to sys.segments table

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

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.timeline;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * DataSegment object plus the overshadowed status for the segment. An immutable object.
+ *
+ * SegmentWithOvershadowedStatus's {@link #compareTo} method considers only the {@link SegmentId}
+ * of the DataSegment object.
+ */
+public class SegmentWithOvershadowedStatus implements Comparable<SegmentWithOvershadowedStatus>
+{
+  private final boolean overshadowed;
+  private final DataSegment dataSegment;
+
+  @JsonCreator
+  public SegmentWithOvershadowedStatus(
+      @JsonProperty("dataSegment") DataSegment dataSegment,
+      @JsonProperty("overshadowed") boolean overshadowed
+  )
+  {
+    this.dataSegment = dataSegment;
+    this.overshadowed = overshadowed;
+  }
+
+  @JsonProperty
+  public boolean isOvershadowed()
+  {
+    return overshadowed;
+  }
+
+  @JsonProperty
+  public DataSegment getDataSegment()
+  {
+    return dataSegment;
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof SegmentWithOvershadowedStatus)) {
+      return false;
+    }
+    final SegmentWithOvershadowedStatus that = (SegmentWithOvershadowedStatus) o;
+    if (!dataSegment.equals(that.dataSegment)) {
+      return false;
+    }
+    if (overshadowed != (that.overshadowed)) {
+      return false;
+    }
+    return true;
+  }
+
+  @Override
+  public int hashCode()
+  {
+    int result = dataSegment.hashCode();
+    result = 31 * result + Boolean.hashCode(overshadowed);
+    return result;
+  }
+
+  @Override
+  public int compareTo(SegmentWithOvershadowedStatus o)
+  {
+    return dataSegment.getId().compareTo(o.dataSegment.getId());
+  }
+}

docs/content/querying/sql.md

@@ -612,6 +612,7 @@ Note that a segment can be served by more than one stream ingestion tasks or His
 |is_published|LONG|Boolean is represented as long type where 1 = true, 0 = false. 1 represents this segment has been published to the metadata store with `used=1`|
 |is_available|LONG|Boolean is represented as long type where 1 = true, 0 = false. 1 if this segment is currently being served by any process(Historical or realtime)|
 |is_realtime|LONG|Boolean is represented as long type where 1 = true, 0 = false. 1 if this segment is being served on any type of realtime tasks|
+|is_overshadowed|LONG|Boolean is represented as long type where 1 = true, 0 = false. 1 if this segment is published and is _fully_ overshadowed by some other published segments. Currently, is_overshadowed is always false for unpublished segments, although this may change in the future. You can filter for segments that "should be published" by filtering for `is_published = 1 AND is_overshadowed = 0`. Segments can briefly be both published and overshadowed if they were recently replaced, but have not been unpublished yet.
 |payload|STRING|JSON-serialized data segment payload|
 
 For example to retrieve all segments for datasource "wikipedia", use the query:

server/src/main/java/org/apache/druid/client/ImmutableDruidDataSource.java

@@ -25,12 +25,17 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSortedMap;
+import com.google.common.collect.Ordering;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.VersionedIntervalTimeline;
 
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 
 /**
  * An immutable collection of metadata of segments ({@link DataSegment} objects), belonging to a particular data source.
@@ -109,6 +114,41 @@ public long getTotalSizeOfSegments()
     return totalSizeOfSegments;
   }
 
+  /**
+   * This method finds the overshadowed segments from the given segments
+   *
+   * @return set of overshadowed segments
+   */
+  public static Set<DataSegment> determineOvershadowedSegments(Iterable<DataSegment> segments)
+  {
+    final Map<String, VersionedIntervalTimeline<String, DataSegment>> timelines = buildTimelines(segments);
+
+    final Set<DataSegment> overshadowedSegments = new HashSet<>();
+    for (DataSegment dataSegment : segments) {
+      final VersionedIntervalTimeline<String, DataSegment> timeline = timelines.get(dataSegment.getDataSource());
+      if (timeline != null && timeline.isOvershadowed(dataSegment.getInterval(), dataSegment.getVersion())) {
+        overshadowedSegments.add(dataSegment);
+      }
+    }
+    return overshadowedSegments;
+  }
+
+  /**
+   * Builds a timeline from given segments
+   *
+   * @return map of datasource to VersionedIntervalTimeline of segments
+   */
+  private static Map<String, VersionedIntervalTimeline<String, DataSegment>> buildTimelines(
+      Iterable<DataSegment> segments
+  )
+  {
+    final Map<String, VersionedIntervalTimeline<String, DataSegment>> timelines = new HashMap<>();
+    segments.forEach(segment -> timelines
+        .computeIfAbsent(segment.getDataSource(), dataSource -> new VersionedIntervalTimeline<>(Ordering.natural()))
+        .add(segment.getInterval(), segment.getVersion(), segment.getShardSpec().createChunk(segment)));
+    return timelines;
+  }
+
   @Override
   public String toString()
   {

server/src/main/java/org/apache/druid/server/coordinator/helper/DruidCoordinatorRuleRunner.java

@@ -20,7 +20,7 @@
 package org.apache.druid.server.coordinator.helper;
 
 import com.google.common.collect.Lists;
-import com.google.common.collect.Ordering;
+import org.apache.druid.client.ImmutableDruidDataSource;
 import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.emitter.EmittingLogger;
 import org.apache.druid.metadata.MetadataRuleManager;
@@ -32,14 +32,9 @@
 import org.apache.druid.server.coordinator.rules.Rule;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.timeline.SegmentId;
-import org.apache.druid.timeline.TimelineObjectHolder;
-import org.apache.druid.timeline.VersionedIntervalTimeline;
 import org.joda.time.DateTime;
 
-import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 
 /**
@@ -89,7 +84,8 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
     // find available segments which are not overshadowed by other segments in DB
     // only those would need to be loaded/dropped
     // anything overshadowed by served segments is dropped automatically by DruidCoordinatorCleanupOvershadowed
-    Set<DataSegment> overshadowed = determineOvershadowedSegments(params);
+    final Set<DataSegment> overshadowed = ImmutableDruidDataSource
+        .determineOvershadowedSegments(params.getAvailableSegments());
 
     for (String tier : cluster.getTierNames()) {
       replicatorThrottler.updateReplicationState(tier);
@@ -138,24 +134,4 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
 
     return params.buildFromExisting().withCoordinatorStats(stats).build();
   }
-
-  private Set<DataSegment> determineOvershadowedSegments(DruidCoordinatorRuntimeParams params)
-  {
-    Map<String, VersionedIntervalTimeline<String, DataSegment>> timelines = new HashMap<>();
-    for (DataSegment segment : params.getAvailableSegments()) {
-      timelines
-          .computeIfAbsent(segment.getDataSource(), dataSource -> new VersionedIntervalTimeline<>(Ordering.natural()))
-          .add(segment.getInterval(), segment.getVersion(), segment.getShardSpec().createChunk(segment));
-    }
-
-    Set<DataSegment> overshadowed = new HashSet<>();
-    for (VersionedIntervalTimeline<String, DataSegment> timeline : timelines.values()) {
-      for (TimelineObjectHolder<String, DataSegment> holder : timeline.findOvershadowed()) {
-        for (DataSegment dataSegment : holder.getObject().payloads()) {
-          overshadowed.add(dataSegment);
-        }
-      }
-    }
-    return overshadowed;
-  }
 }

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

@@ -37,6 +37,7 @@
 import org.apache.druid.server.security.ResourceAction;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.SegmentWithOvershadowedStatus;
 import org.joda.time.Interval;
 
 import javax.servlet.http.HttpServletRequest;
@@ -51,6 +52,7 @@
 import javax.ws.rs.core.Response;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
@@ -147,7 +149,8 @@ public Response getDatabaseSegmentDataSource(@PathParam("dataSourceName") final
   @Produces(MediaType.APPLICATION_JSON)
   public Response getDatabaseSegments(
       @Context final HttpServletRequest req,
-      @QueryParam("datasources") final Set<String> datasources
+      @QueryParam("datasources") final Set<String> datasources,
+      @QueryParam("includeOvershadowedStatus") final String includeOvershadowedStatus
   )
   {
     // If we haven't polled the metadata store yet, use an empty list of datasources.
@@ -159,14 +162,61 @@ public Response getDatabaseSegments(
     }
     final Stream<DataSegment> metadataSegments = dataSourceStream.flatMap(t -> t.getSegments().stream());
 
-    final Function<DataSegment, Iterable<ResourceAction>> raGenerator = segment -> Collections.singletonList(
-        AuthorizationUtils.DATASOURCE_READ_RA_GENERATOR.apply(segment.getDataSource()));
+    if (includeOvershadowedStatus != null) {
+      final Iterable<SegmentWithOvershadowedStatus> authorizedSegments = findAuthorizedSegmentWithOvershadowedStatus(
+          req,
+          druidDataSources,
+          metadataSegments
+      );
+      Response.ResponseBuilder builder = Response.status(Response.Status.OK);
+      return builder.entity(authorizedSegments).build();
+    } else {
+
+      final Function<DataSegment, Iterable<ResourceAction>> raGenerator = segment -> Collections.singletonList(
+          AuthorizationUtils.DATASOURCE_READ_RA_GENERATOR.apply(segment.getDataSource()));
+
+      final Iterable<DataSegment> authorizedSegments = AuthorizationUtils.filterAuthorizedResources(
+          req,
+          metadataSegments::iterator,
+          raGenerator,
+          authorizerMapper
+      );
+
+      Response.ResponseBuilder builder = Response.status(Response.Status.OK);
+      return builder.entity(authorizedSegments).build();
+    }
+  }
 
-    final Iterable<DataSegment> authorizedSegments =
-        AuthorizationUtils.filterAuthorizedResources(req, metadataSegments::iterator, raGenerator, authorizerMapper);
+  private Iterable<SegmentWithOvershadowedStatus> findAuthorizedSegmentWithOvershadowedStatus(
+      HttpServletRequest req,
+      Collection<ImmutableDruidDataSource> druidDataSources,
+      Stream<DataSegment> metadataSegments
+  )
+  {
+    // It's fine to add all overshadowed segments to a single collection because only
+    // a small fraction of the segments in the cluster are expected to be overshadowed,
+    // so building this collection shouldn't generate a lot of garbage.
+    final Set<DataSegment> overshadowedSegments = new HashSet<>();
+    for (ImmutableDruidDataSource dataSource : druidDataSources) {
+      overshadowedSegments.addAll(ImmutableDruidDataSource.determineOvershadowedSegments(dataSource.getSegments()));
+    }
 
-    final Response.ResponseBuilder builder = Response.status(Response.Status.OK);
-    return builder.entity(authorizedSegments).build();
+    final Stream<SegmentWithOvershadowedStatus> segmentsWithOvershadowedStatus = metadataSegments
+        .map(segment -> new SegmentWithOvershadowedStatus(
+            segment,
+            overshadowedSegments.contains(segment)
+        ));
+
+    final Function<SegmentWithOvershadowedStatus, Iterable<ResourceAction>> raGenerator = segment -> Collections
+        .singletonList(AuthorizationUtils.DATASOURCE_READ_RA_GENERATOR.apply(segment.getDataSegment().getDataSource()));
+
+    final Iterable<SegmentWithOvershadowedStatus> authorizedSegments = AuthorizationUtils.filterAuthorizedResources(
+        req,
+        segmentsWithOvershadowedStatus::iterator,
+        raGenerator,
+        authorizerMapper
+    );
+    return authorizedSegments;
   }
 
   @GET