diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/MarkSegmentsAsUnusedAction.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/MarkSegmentsAsUnusedAction.java
index ddf57afbc185..93cb75280fac 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/MarkSegmentsAsUnusedAction.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/MarkSegmentsAsUnusedAction.java
@@ -63,9 +63,8 @@ public TypeReference<Integer> getReturnTypeReference()
   @Override
   public Integer perform(Task task, TaskActionToolbox toolbox)
   {
-    int numMarked = toolbox.getIndexerMetadataStorageCoordinator()
-                           .markSegmentsAsUnusedWithinInterval(dataSource, interval);
-    return numMarked;
+    return toolbox.getIndexerMetadataStorageCoordinator()
+                  .markSegmentsAsUnusedWithinInterval(dataSource, interval);
   }
 
   @Override
diff --git a/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManager.java b/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManager.java
index 540eba990f22..91e33b7a231c 100644
--- a/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManager.java
+++ b/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManager.java
@@ -53,7 +53,20 @@ public interface SegmentsMetadataManager
    */
   int markAsUsedAllNonOvershadowedSegmentsInDataSource(String dataSource);
 
-  int markAsUsedNonOvershadowedSegmentsInInterval(String dataSource, Interval interval);
+  /**
+   * Marks non-overshadowed unused segments for the given interval as used.
+   * @return Number of segments updated
+   */
+  default int markAsUsedNonOvershadowedSegmentsInInterval(String dataSource, Interval interval)
+  {
+    return markAsUsedNonOvershadowedSegmentsInInterval(dataSource, interval, null);
+  }
+
+  /**
+   * Marks non-overshadowed unused segments for the given interval and optional list of versions as used.
+   * @return Number of segments updated
+   */
+  int markAsUsedNonOvershadowedSegmentsInInterval(String dataSource, Interval interval, List<String> versions);
 
   /**
    * Marks the given segment IDs as "used" only if there are not already overshadowed
@@ -81,7 +94,22 @@ public interface SegmentsMetadataManager
    */
   int markAsUnusedAllSegmentsInDataSource(String dataSource);
 
-  int markAsUnusedSegmentsInInterval(String dataSource, Interval interval);
+  /**
+   * Marks segments as unused that are fully contained in the specified interval. Segments that are already marked as
+   * unused are not updated.
+   * @return Number of segments updated
+   */
+  default int markAsUnusedSegmentsInInterval(String dataSource, Interval interval)
+  {
+    return markAsUnusedSegmentsInInterval(dataSource, interval, null);
+  }
+
+  /**
+   * Marks segments as unused that are fully contained in the specified interval with an optional list of versions.
+   * Segments that are already marked as unused are not updated.
+   * @return The number of segments updated
+   */
+  int markAsUnusedSegmentsInInterval(String dataSource, Interval interval, List<String> versions);
 
   int markSegmentsAsUnused(Set<SegmentId> segmentIds);
 
diff --git a/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataManager.java b/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataManager.java
index 4a268a2257da..66a60e072c02 100644
--- a/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataManager.java
+++ b/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataManager.java
@@ -654,21 +654,25 @@ public boolean markSegmentAsUsed(final String segmentId)
   @Override
   public int markAsUsedAllNonOvershadowedSegmentsInDataSource(final String dataSource)
   {
-    return doMarkAsUsedNonOvershadowedSegments(dataSource, null);
+    return doMarkAsUsedNonOvershadowedSegments(dataSource, null, null);
   }
 
   @Override
-  public int markAsUsedNonOvershadowedSegmentsInInterval(final String dataSource, final Interval interval)
+  public int markAsUsedNonOvershadowedSegmentsInInterval(
+      final String dataSource,
+      final Interval interval,
+      @Nullable final List<String> versions
+  )
   {
     Preconditions.checkNotNull(interval);
-    return doMarkAsUsedNonOvershadowedSegments(dataSource, interval);
+    return doMarkAsUsedNonOvershadowedSegments(dataSource, interval, versions);
   }
 
-  /**
-   * Implementation for both {@link #markAsUsedAllNonOvershadowedSegmentsInDataSource} (if the given interval is null)
-   * and {@link #markAsUsedNonOvershadowedSegmentsInInterval}.
-   */
-  private int doMarkAsUsedNonOvershadowedSegments(String dataSourceName, @Nullable Interval interval)
+  private int doMarkAsUsedNonOvershadowedSegments(
+      final String dataSourceName,
+      final @Nullable Interval interval,
+      final @Nullable List<String> versions
+  )
   {
     final List<DataSegment> unusedSegments = new ArrayList<>();
     final SegmentTimeline timeline = new SegmentTimeline();
@@ -682,12 +686,12 @@ private int doMarkAsUsedNonOvershadowedSegments(String dataSourceName, @Nullable
               interval == null ? Intervals.ONLY_ETERNITY : Collections.singletonList(interval);
 
           try (final CloseableIterator<DataSegment> iterator =
-                   queryTool.retrieveUsedSegments(dataSourceName, intervals)) {
+                   queryTool.retrieveUsedSegments(dataSourceName, intervals, versions)) {
             timeline.addSegments(iterator);
           }
 
           try (final CloseableIterator<DataSegment> iterator =
-                   queryTool.retrieveUnusedSegments(dataSourceName, intervals, null, null, null, null, null)) {
+                   queryTool.retrieveUnusedSegments(dataSourceName, intervals, versions, null, null, null, null)) {
             while (iterator.hasNext()) {
               final DataSegment dataSegment = iterator.next();
               timeline.addSegments(Iterators.singletonIterator(dataSegment));
@@ -796,7 +800,7 @@ private CloseableIterator<DataSegment> retrieveUsedSegmentsOverlappingIntervals(
   private int markSegmentsAsUsed(final List<SegmentId> segmentIds)
   {
     if (segmentIds.isEmpty()) {
-      log.info("No segments found to update!");
+      log.info("No segments found to mark as used.");
       return 0;
     }
 
@@ -856,13 +860,18 @@ public int markSegmentsAsUnused(Set<SegmentId> segmentIds)
   }
 
   @Override
-  public int markAsUnusedSegmentsInInterval(String dataSourceName, Interval interval)
+  public int markAsUnusedSegmentsInInterval(
+      final String dataSource,
+      final Interval interval,
+      @Nullable final List<String> versions
+  )
   {
+    Preconditions.checkNotNull(interval);
     try {
       return connector.getDBI().withHandle(
           handle ->
               SqlSegmentsMetadataQuery.forHandle(handle, connector, dbTables.get(), jsonMapper)
-                                      .markSegmentsUnused(dataSourceName, interval)
+                                      .markSegmentsUnused(dataSource, interval, versions)
       );
     }
     catch (Exception e) {
diff --git a/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java b/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java
index 29465cd665ba..49f6a7de5734 100644
--- a/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java
+++ b/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java
@@ -119,11 +119,24 @@ public CloseableIterator<DataSegment> retrieveUsedSegments(
       final String dataSource,
       final Collection<Interval> intervals
   )
+  {
+    return retrieveUsedSegments(dataSource, intervals, null);
+  }
+
+  /**
+   * Similar to {@link #retrieveUsedSegments}, but with an additional {@code versions} argument. When {@code versions}
+   * is specified, all used segments in the specified {@code intervals} and {@code versions} are retrieved.
+   */
+  public CloseableIterator<DataSegment> retrieveUsedSegments(
+      final String dataSource,
+      final Collection<Interval> intervals,
+      final List<String> versions
+  )
   {
     return retrieveSegments(
         dataSource,
         intervals,
-        null,
+        versions,
         IntervalMode.OVERLAPS,
         true,
         null,
@@ -314,19 +327,34 @@ public int markSegments(final Collection<SegmentId> segmentIds, final boolean us
   /**
    * Marks all used segments that are *fully contained by* a particular interval as unused.
    *
-   * @return the number of segments actually modified.
+   * @return Number of segments updated.
    */
   public int markSegmentsUnused(final String dataSource, final Interval interval)
+  {
+    return markSegmentsUnused(dataSource, interval, null);
+  }
+
+  /**
+   * Marks all used segments that are *fully contained by* a particular interval filtered by an optional list of versions
+   * as unused.
+   *
+   * @return Number of segments updated.
+   */
+  public int markSegmentsUnused(final String dataSource, final Interval interval, @Nullable final List<String> versions)
   {
     if (Intervals.isEternity(interval)) {
-      return handle
-          .createStatement(
-              StringUtils.format(
-                  "UPDATE %s SET used=:used, used_status_last_updated = :used_status_last_updated "
-                  + "WHERE dataSource = :dataSource AND used = true",
-                  dbTables.getSegmentsTable()
-              )
+      final StringBuilder sb = new StringBuilder();
+      sb.append(
+          StringUtils.format(
+              "UPDATE %s SET used=:used, used_status_last_updated = :used_status_last_updated "
+              + "WHERE dataSource = :dataSource AND used = true",
+              dbTables.getSegmentsTable()
           )
+      );
+      appendConditionForVersions(sb, versions);
+
+      return handle
+          .createStatement(sb.toString())
           .bind("dataSource", dataSource)
           .bind("used", false)
           .bind("used_status_last_updated", DateTimes.nowUtc().toString())
@@ -336,15 +364,19 @@ public int markSegmentsUnused(final String dataSource, final Interval interval)
       // Safe to write a WHERE clause with this interval. Note that it is unsafe if the years are different, because
       // that means extra characters can sneak in. (Consider a query interval like "2000-01-01/2001-01-01" and a
       // segment interval like "20001/20002".)
-      return handle
-          .createStatement(
-              StringUtils.format(
-                  "UPDATE %s SET used=:used, used_status_last_updated = :used_status_last_updated "
-                  + "WHERE dataSource = :dataSource AND used = true AND %s",
-                  dbTables.getSegmentsTable(),
-                  IntervalMode.CONTAINS.makeSqlCondition(connector.getQuoteString(), ":start", ":end")
-              )
+      final StringBuilder sb = new StringBuilder();
+      sb.append(
+          StringUtils.format(
+              "UPDATE %s SET used=:used, used_status_last_updated = :used_status_last_updated "
+              + "WHERE dataSource = :dataSource AND used = true AND %s",
+              dbTables.getSegmentsTable(),
+              IntervalMode.CONTAINS.makeSqlCondition(connector.getQuoteString(), ":start", ":end")
           )
+      );
+      appendConditionForVersions(sb, versions);
+
+      return handle
+          .createStatement(sb.toString())
           .bind("dataSource", dataSource)
           .bind("used", false)
           .bind("start", interval.getStart().toString())
@@ -358,7 +390,7 @@ public int markSegmentsUnused(final String dataSource, final Interval interval)
               retrieveSegments(
                   dataSource,
                   Collections.singletonList(interval),
-                  null,
+                  versions,
                   IntervalMode.CONTAINS,
                   true,
                   null,
@@ -680,12 +712,7 @@ private Query<Map<String, Object>> buildSegmentsTableQuery(
       appendConditionForIntervalsAndMatchMode(sb, intervals, matchMode, connector);
     }
 
-    if (!CollectionUtils.isNullOrEmpty(versions)) {
-      final String versionsStr = versions.stream()
-                                          .map(version -> "'" + version + "'")
-                                          .collect(Collectors.joining(","));
-      sb.append(StringUtils.format(" AND version IN (%s)", versionsStr));
-    }
+    appendConditionForVersions(sb, versions);
 
     // Add the used_status_last_updated time filter only for unused segments when maxUsedStatusLastUpdatedTime is non-null.
     final boolean addMaxUsedLastUpdatedTimeFilter = !used && maxUsedStatusLastUpdatedTime != null;
@@ -834,6 +861,21 @@ private static int computeNumChangedSegments(List<String> segmentIds, int[] segm
     return numChangedSegments;
   }
 
+  private static void appendConditionForVersions(
+      final StringBuilder sb,
+      final List<String> versions
+  )
+  {
+    if (CollectionUtils.isNullOrEmpty(versions)) {
+      return;
+    }
+
+    final String versionsCsv = versions.stream()
+                                       .map(version -> "'" + version + "'")
+                                       .collect(Collectors.joining(","));
+    sb.append(StringUtils.format(" AND version IN (%s)", versionsCsv));
+  }
+
   enum IntervalMode
   {
     CONTAINS {
diff --git a/server/src/main/java/org/apache/druid/server/http/DataSourcesResource.java b/server/src/main/java/org/apache/druid/server/http/DataSourcesResource.java
index d0458243fcfc..921ac02b4ffb 100644
--- a/server/src/main/java/org/apache/druid/server/http/DataSourcesResource.java
+++ b/server/src/main/java/org/apache/druid/server/http/DataSourcesResource.java
@@ -217,10 +217,14 @@ public Response markAsUsedNonOvershadowedSegments(
           .build();
     } else {
       SegmentUpdateOperation operation = () -> {
-
         final Interval interval = payload.getInterval();
+        final List<String> versions = payload.getVersions();
         if (interval != null) {
-          return segmentsMetadataManager.markAsUsedNonOvershadowedSegmentsInInterval(dataSourceName, interval);
+          if (versions != null) {
+            return segmentsMetadataManager.markAsUsedNonOvershadowedSegmentsInInterval(dataSourceName, interval, versions);
+          } else {
+            return segmentsMetadataManager.markAsUsedNonOvershadowedSegmentsInInterval(dataSourceName, interval);
+          }
         } else {
           final Set<String> segmentIds = payload.getSegmentIds();
           if (segmentIds == null || segmentIds.isEmpty()) {
@@ -266,9 +270,14 @@ public Response markSegmentsAsUnused(
     } else {
       SegmentUpdateOperation operation = () -> {
         final Interval interval = payload.getInterval();
+        final List<String> versions = payload.getVersions();
         final int numUpdatedSegments;
         if (interval != null) {
-          numUpdatedSegments = segmentsMetadataManager.markAsUnusedSegmentsInInterval(dataSourceName, interval);
+          if (versions != null) {
+            numUpdatedSegments = segmentsMetadataManager.markAsUnusedSegmentsInInterval(dataSourceName, interval, versions);
+          } else {
+            numUpdatedSegments = segmentsMetadataManager.markAsUnusedSegmentsInInterval(dataSourceName, interval);
+          }
         } else {
           final Set<SegmentId> segmentIds =
               payload.getSegmentIds()
@@ -995,15 +1004,18 @@ protected static class MarkDataSourceSegmentsPayload
   {
     private final Interval interval;
     private final Set<String> segmentIds;
+    private final List<String> versions;
 
     @JsonCreator
     public MarkDataSourceSegmentsPayload(
         @JsonProperty("interval") Interval interval,
-        @JsonProperty("segmentIds") Set<String> segmentIds
+        @JsonProperty("segmentIds") Set<String> segmentIds,
+        @JsonProperty("versions") List<String> versions
     )
     {
       this.interval = interval;
       this.segmentIds = segmentIds;
+      this.versions = versions;
     }
 
     @JsonProperty
@@ -1018,9 +1030,15 @@ public Set<String> getSegmentIds()
       return segmentIds;
     }
 
+    @JsonProperty
+    public List<String> getVersions()
+    {
+      return versions;
+    }
+
     public boolean isValid()
     {
-      return (interval == null ^ segmentIds == null) && (segmentIds == null || !segmentIds.isEmpty());
+      return (interval == null ^ segmentIds == null) && (segmentIds == null || !segmentIds.isEmpty()); // fixme
     }
   }
 }
diff --git a/server/src/test/java/org/apache/druid/metadata/SqlSegmentsMetadataManagerTest.java b/server/src/test/java/org/apache/druid/metadata/SqlSegmentsMetadataManagerTest.java
index b177b40c5876..f00382003aae 100644
--- a/server/src/test/java/org/apache/druid/metadata/SqlSegmentsMetadataManagerTest.java
+++ b/server/src/test/java/org/apache/druid/metadata/SqlSegmentsMetadataManagerTest.java
@@ -41,6 +41,7 @@
 import org.apache.druid.timeline.SegmentId;
 import org.apache.druid.timeline.partition.NoneShardSpec;
 import org.hamcrest.MatcherAssert;
+import org.joda.time.DateTime;
 import org.joda.time.Duration;
 import org.joda.time.Interval;
 import org.joda.time.Period;
@@ -589,6 +590,156 @@ public void testMarkAsUsedNonOvershadowedSegments() throws Exception
     );
   }
 
+  @Test
+  public void testMarkAsUsedNonOvershadowedSegmentsInEternityIntervalWithVersions() throws Exception
+  {
+    publishWikiSegments();
+    sqlSegmentsMetadataManager.startPollingDatabasePeriodically();
+    sqlSegmentsMetadataManager.poll();
+    Assert.assertTrue(sqlSegmentsMetadataManager.isPollingDatabasePeriodically());
+
+    final DataSegment koalaSegment1 = createSegment(
+        DS.KOALA,
+        "2017-10-15T00:00:00.000/2017-10-17T00:00:00.000",
+        "2017-10-15T20:19:12.565Z"
+    );
+
+    final DataSegment koalaSegment2 = createSegment(
+        DS.KOALA,
+        "2017-10-17T00:00:00.000/2017-10-18T00:00:00.000",
+        "2017-10-16T20:19:12.565Z"
+    );
+
+    // Overshadowed by koalaSegment2
+    final DataSegment koalaSegment3 = createSegment(
+        DS.KOALA,
+        "2017-10-17T00:00:00.000/2017-10-18T00:00:00.000",
+        "2017-10-15T20:19:12.565Z"
+    );
+
+    publishUnusedSegments(koalaSegment1, koalaSegment2, koalaSegment3);
+
+    sqlSegmentsMetadataManager.poll();
+    Assert.assertEquals(
+        ImmutableSet.of(wikiSegment1, wikiSegment2),
+        ImmutableSet.copyOf(sqlSegmentsMetadataManager.iterateAllUsedSegments())
+    );
+    Assert.assertEquals(
+        2,
+        sqlSegmentsMetadataManager.markAsUsedNonOvershadowedSegmentsInInterval(
+            DS.KOALA,
+            Intervals.ETERNITY,
+            ImmutableList.of("2017-10-15T20:19:12.565Z", "2017-10-16T20:19:12.565Z")
+        )
+    );
+    sqlSegmentsMetadataManager.poll();
+
+    Assert.assertEquals(
+        ImmutableSet.of(wikiSegment1, wikiSegment2, koalaSegment1, koalaSegment2),
+        ImmutableSet.copyOf(sqlSegmentsMetadataManager.iterateAllUsedSegments())
+    );
+  }
+
+  @Test
+  public void testMarkAsUsedNonOvershadowedSegmentsInFiniteIntervalWithVersions() throws Exception
+  {
+    publishWikiSegments();
+    sqlSegmentsMetadataManager.startPollingDatabasePeriodically();
+    sqlSegmentsMetadataManager.poll();
+    Assert.assertTrue(sqlSegmentsMetadataManager.isPollingDatabasePeriodically());
+
+    final DataSegment koalaSegment1 = createSegment(
+        DS.KOALA,
+        "2017-10-15T00:00:00.000/2017-10-17T00:00:00.000",
+        "2017-10-15T20:19:12.565Z"
+    );
+
+    final DataSegment koalaSegment2 = createSegment(
+        DS.KOALA,
+        "2017-10-17T00:00:00.000/2017-10-18T00:00:00.000",
+        "2017-10-16T20:19:12.565Z"
+    );
+
+    // Overshadowed by koalaSegment2
+    final DataSegment koalaSegment3 = createSegment(
+        DS.KOALA,
+        "2017-10-17T00:00:00.000/2017-10-18T00:00:00.000",
+        "2017-10-15T20:19:12.565Z"
+    );
+
+    publishUnusedSegments(koalaSegment1, koalaSegment2, koalaSegment3);
+
+    sqlSegmentsMetadataManager.poll();
+    Assert.assertEquals(
+        ImmutableSet.of(wikiSegment1, wikiSegment2),
+        ImmutableSet.copyOf(sqlSegmentsMetadataManager.iterateAllUsedSegments())
+    );
+    Assert.assertEquals(
+        2,
+        sqlSegmentsMetadataManager.markAsUsedNonOvershadowedSegmentsInInterval(
+            DS.KOALA,
+            Intervals.of("2017-10-15/2017-10-18"),
+            ImmutableList.of("2017-10-15T20:19:12.565Z", "2017-10-16T20:19:12.565Z")
+        )
+    );
+    sqlSegmentsMetadataManager.poll();
+
+    Assert.assertEquals(
+        ImmutableSet.of(wikiSegment1, wikiSegment2, koalaSegment1, koalaSegment2),
+        ImmutableSet.copyOf(sqlSegmentsMetadataManager.iterateAllUsedSegments())
+    );
+  }
+
+  @Test
+  public void testMarkAsUsedNonOvershadowedSegmentsWithNonExistentVersions() throws Exception
+  {
+    publishWikiSegments();
+    sqlSegmentsMetadataManager.startPollingDatabasePeriodically();
+    sqlSegmentsMetadataManager.poll();
+    Assert.assertTrue(sqlSegmentsMetadataManager.isPollingDatabasePeriodically());
+
+    final DataSegment koalaSegment1 = createSegment(
+        DS.KOALA,
+        "2017-10-15T00:00:00.000/2017-10-17T00:00:00.000",
+        "2017-10-15T20:19:12.565Z"
+    );
+
+    final DataSegment koalaSegment2 = createSegment(
+        DS.KOALA,
+        "2017-10-17T00:00:00.000/2017-10-18T00:00:00.000",
+        "2017-10-16T20:19:12.565Z"
+    );
+
+    // Overshadowed by koalaSegment2
+    final DataSegment koalaSegment3 = createSegment(
+        DS.KOALA,
+        "2017-10-17T00:00:00.000/2017-10-18T00:00:00.000",
+        "2017-10-15T20:19:12.565Z"
+    );
+
+    publishUnusedSegments(koalaSegment1, koalaSegment2, koalaSegment3);
+
+    sqlSegmentsMetadataManager.poll();
+    Assert.assertEquals(
+        ImmutableSet.of(wikiSegment1, wikiSegment2),
+        ImmutableSet.copyOf(sqlSegmentsMetadataManager.iterateAllUsedSegments())
+    );
+    Assert.assertEquals(
+        0,
+        sqlSegmentsMetadataManager.markAsUsedNonOvershadowedSegmentsInInterval(
+            DS.KOALA,
+            Intervals.ETERNITY,
+            ImmutableList.of("foo", "bar")
+        )
+    );
+    sqlSegmentsMetadataManager.poll();
+
+    Assert.assertEquals(
+        ImmutableSet.of(wikiSegment1, wikiSegment2),
+        ImmutableSet.copyOf(sqlSegmentsMetadataManager.iterateAllUsedSegments())
+    );
+  }
+
   @Test
   public void testMarkAsUsedNonOvershadowedSegmentsInvalidDataSource() throws Exception
   {
@@ -796,6 +947,104 @@ public void testMarkAsUnusedSegmentsInInterval() throws IOException
     );
   }
 
+  @Test
+  public void testMarkAsUnusedSegmentsInIntervalAndVersions() throws IOException
+  {
+    publishWikiSegments();
+    sqlSegmentsMetadataManager.startPollingDatabasePeriodically();
+    sqlSegmentsMetadataManager.poll();
+    Assert.assertTrue(sqlSegmentsMetadataManager.isPollingDatabasePeriodically());
+
+    final DateTime now = DateTimes.nowUtc();
+    final String v1 = now.toString();
+    final String v2 = now.plus(Duration.standardDays(1)).toString();
+
+    final DataSegment koalaSegment1 = createSegment(
+        DS.KOALA,
+        "2017-10-15T00:00:00.000/2017-10-16T00:00:00.000",
+        v1
+    );
+    final DataSegment koalaSegment2 = createSegment(
+        DS.KOALA,
+        "2017-10-17T00:00:00.000/2017-10-18T00:00:00.000",
+        v2
+    );
+    final DataSegment koalaSegment3 = createSegment(
+        DS.KOALA,
+        "2017-10-19T00:00:00.000/2017-10-20T00:00:00.000",
+        v2
+    );
+
+    publisher.publishSegment(koalaSegment1);
+    publisher.publishSegment(koalaSegment2);
+    publisher.publishSegment(koalaSegment3);
+    final Interval theInterval = Intervals.of("2017-10-15/2017-10-18");
+
+    Assert.assertEquals(
+        2,
+        sqlSegmentsMetadataManager.markAsUnusedSegmentsInInterval(
+            DS.KOALA,
+            theInterval,
+            ImmutableList.of(v1, v2)
+        )
+    );
+
+    sqlSegmentsMetadataManager.poll();
+    Assert.assertEquals(
+        ImmutableSet.of(wikiSegment1, wikiSegment2, koalaSegment3),
+        ImmutableSet.copyOf(sqlSegmentsMetadataManager.iterateAllUsedSegments())
+    );
+  }
+
+  @Test
+  public void testMarkAsUnusedSegmentsInIntervalAndNonExistentVersions() throws IOException
+  {
+    publishWikiSegments();
+    sqlSegmentsMetadataManager.startPollingDatabasePeriodically();
+    sqlSegmentsMetadataManager.poll();
+    Assert.assertTrue(sqlSegmentsMetadataManager.isPollingDatabasePeriodically());
+
+    final DateTime now = DateTimes.nowUtc();
+    final String v1 = now.toString();
+    final String v2 = now.plus(Duration.standardDays(1)).toString();
+
+    final DataSegment koalaSegment1 = createSegment(
+        DS.KOALA,
+        "2017-10-15T00:00:00.000/2017-10-16T00:00:00.000",
+        v1
+    );
+    final DataSegment koalaSegment2 = createSegment(
+        DS.KOALA,
+        "2017-10-17T00:00:00.000/2017-10-18T00:00:00.000",
+        v2
+    );
+    final DataSegment koalaSegment3 = createSegment(
+        DS.KOALA,
+        "2017-10-19T00:00:00.000/2017-10-20T00:00:00.000",
+        v2
+    );
+
+    publisher.publishSegment(koalaSegment1);
+    publisher.publishSegment(koalaSegment2);
+    publisher.publishSegment(koalaSegment3);
+    final Interval theInterval = Intervals.of("2017-10-15/2017-10-18");
+
+    Assert.assertEquals(
+        0,
+        sqlSegmentsMetadataManager.markAsUnusedSegmentsInInterval(
+            DS.KOALA,
+            theInterval,
+            ImmutableList.of("foo", "bar", "baz")
+        )
+    );
+
+    sqlSegmentsMetadataManager.poll();
+    Assert.assertEquals(
+        ImmutableSet.of(wikiSegment1, wikiSegment2, koalaSegment1, koalaSegment2, koalaSegment3),
+        ImmutableSet.copyOf(sqlSegmentsMetadataManager.iterateAllUsedSegments())
+    );
+  }
+
   @Test
   public void testMarkAsUnusedSegmentsInIntervalWithOverlappingInterval() throws IOException
   {
diff --git a/server/src/test/java/org/apache/druid/metadata/TestDerbyConnector.java b/server/src/test/java/org/apache/druid/metadata/TestDerbyConnector.java
index 1fa2ef2302c9..fc2330cf1f0e 100644
--- a/server/src/test/java/org/apache/druid/metadata/TestDerbyConnector.java
+++ b/server/src/test/java/org/apache/druid/metadata/TestDerbyConnector.java
@@ -145,7 +145,7 @@ public SegmentsTable segments()
   }
 
   /**
-   * A wrapper class for queries on the segments table.
+   * A wrapper class for updating the segments table.
    */
   public static class SegmentsTable
   {
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/simulate/TestSegmentsMetadataManager.java b/server/src/test/java/org/apache/druid/server/coordinator/simulate/TestSegmentsMetadataManager.java
index adf12ae70543..356976aa3a27 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/simulate/TestSegmentsMetadataManager.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/simulate/TestSegmentsMetadataManager.java
@@ -87,7 +87,7 @@ public int markAsUsedAllNonOvershadowedSegmentsInDataSource(String dataSource)
   }
 
   @Override
-  public int markAsUsedNonOvershadowedSegmentsInInterval(String dataSource, Interval interval)
+  public int markAsUsedNonOvershadowedSegmentsInInterval(String dataSource, Interval interval, List<String> versions)
   {
     return 0;
   }
@@ -116,7 +116,7 @@ public int markAsUnusedAllSegmentsInDataSource(String dataSource)
   }
 
   @Override
-  public int markAsUnusedSegmentsInInterval(String dataSource, Interval interval)
+  public int markAsUnusedSegmentsInInterval(String dataSource, Interval interval, List<String> versions)
   {
     return 0;
   }
diff --git a/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java b/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java
index ace68ad3c21e..a74d93920661 100644
--- a/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java
+++ b/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java
@@ -742,7 +742,7 @@ public void testMarkAsUsedNonOvershadowedSegmentsInterval()
     DataSourcesResource dataSourcesResource = createResource();
     Response response = dataSourcesResource.markAsUsedNonOvershadowedSegments(
         "datasource1",
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(interval, null)
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(interval, null, null)
     );
     Assert.assertEquals(200, response.getStatus());
     EasyMock.verify(segmentsMetadataManager, inventoryView, server);
@@ -761,7 +761,7 @@ public void testMarkAsUsedNonOvershadowedSegmentsIntervalNoneUpdated()
 
     Response response = dataSourcesResource.markAsUsedNonOvershadowedSegments(
         "datasource1",
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(interval, null)
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(interval, null, null)
     );
     Assert.assertEquals(ImmutableMap.of("numChangedSegments", 0), response.getEntity());
     EasyMock.verify(segmentsMetadataManager, inventoryView, server);
@@ -780,7 +780,7 @@ public void testMarkAsUsedNonOvershadowedSegmentsSet()
 
     Response response = dataSourcesResource.markAsUsedNonOvershadowedSegments(
         "datasource1",
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(null, segmentIds)
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(null, segmentIds, null)
     );
     Assert.assertEquals(200, response.getStatus());
     EasyMock.verify(segmentsMetadataManager, inventoryView, server);
@@ -799,7 +799,7 @@ public void testMarkAsUsedNonOvershadowedSegmentsIntervalException()
 
     Response response = dataSourcesResource.markAsUsedNonOvershadowedSegments(
         "datasource1",
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(interval, null)
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(interval, null, null)
     );
     Assert.assertEquals(500, response.getStatus());
     EasyMock.verify(segmentsMetadataManager, inventoryView, server);
@@ -818,7 +818,7 @@ public void testMarkAsUsedNonOvershadowedSegmentsNoDataSource()
 
     Response response = dataSourcesResource.markAsUsedNonOvershadowedSegments(
         "datasource1",
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(Intervals.of("2010-01-22/P1D"), null)
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(Intervals.of("2010-01-22/P1D"), null, null)
     );
     Assert.assertEquals(200, response.getStatus());
     Assert.assertEquals(ImmutableMap.of("numChangedSegments", 0), response.getEntity());
@@ -832,7 +832,7 @@ public void testMarkAsUsedNonOvershadowedSegmentsInvalidPayloadNoArguments()
 
     Response response = dataSourcesResource.markAsUsedNonOvershadowedSegments(
         "datasource1",
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(null, null)
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(null, null, null)
     );
     Assert.assertEquals(400, response.getStatus());
   }
@@ -844,7 +844,7 @@ public void testMarkAsUsedNonOvershadowedSegmentsInvalidPayloadBothArguments()
 
     Response response = dataSourcesResource.markAsUsedNonOvershadowedSegments(
         "datasource1",
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(Intervals.of("2010-01-22/P1D"), ImmutableSet.of())
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(Intervals.of("2010-01-22/P1D"), ImmutableSet.of(), null)
     );
     Assert.assertEquals(400, response.getStatus());
   }
@@ -856,7 +856,7 @@ public void testMarkAsUsedNonOvershadowedSegmentsInvalidPayloadEmptyArray()
 
     Response response = dataSourcesResource.markAsUsedNonOvershadowedSegments(
         "datasource1",
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(null, ImmutableSet.of())
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(null, ImmutableSet.of(), null)
     );
     Assert.assertEquals(400, response.getStatus());
   }
@@ -1018,7 +1018,8 @@ public void testMarkSegmentsAsUnused()
             null,
             segmentIds.stream()
                       .map(SegmentId::toString)
-                      .collect(Collectors.toSet())
+                      .collect(Collectors.toSet()),
+            null
         );
 
     DataSourcesResource dataSourcesResource = createResource();
@@ -1047,7 +1048,8 @@ public void testMarkSegmentsAsUnusedNoChanges()
             null,
             segmentIds.stream()
                       .map(SegmentId::toString)
-                      .collect(Collectors.toSet())
+                      .collect(Collectors.toSet()),
+            null
         );
 
     DataSourcesResource dataSourcesResource = createResource();
@@ -1078,7 +1080,8 @@ public void testMarkSegmentsAsUnusedException()
             null,
             segmentIds.stream()
                       .map(SegmentId::toString)
-                      .collect(Collectors.toSet())
+                      .collect(Collectors.toSet()),
+            null
         );
 
     DataSourcesResource dataSourcesResource =
@@ -1098,7 +1101,7 @@ public void testMarkAsUnusedSegmentsInInterval()
     EasyMock.replay(segmentsMetadataManager, inventoryView, server);
 
     final DataSourcesResource.MarkDataSourceSegmentsPayload payload =
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(theInterval, null);
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(theInterval, null, null);
 
     DataSourcesResource dataSourcesResource = createResource();
     prepareRequestForAudit();
@@ -1118,7 +1121,7 @@ public void testMarkAsUnusedSegmentsInIntervalNoChanges()
     EasyMock.replay(segmentsMetadataManager, inventoryView, server);
 
     final DataSourcesResource.MarkDataSourceSegmentsPayload payload =
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(theInterval, null);
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(theInterval, null, null);
 
     DataSourcesResource dataSourcesResource = createResource();
     prepareRequestForAudit();
@@ -1139,7 +1142,7 @@ public void testMarkAsUnusedSegmentsInIntervalException()
     EasyMock.replay(segmentsMetadataManager, inventoryView, server);
 
     final DataSourcesResource.MarkDataSourceSegmentsPayload payload =
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(theInterval, null);
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(theInterval, null, null);
 
     DataSourcesResource dataSourcesResource =
         createResource();
@@ -1157,7 +1160,7 @@ public void testMarkAsUnusedSegmentsInIntervalNoDataSource()
     EasyMock.replay(segmentsMetadataManager, inventoryView, server);
 
     final DataSourcesResource.MarkDataSourceSegmentsPayload payload =
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(theInterval, null);
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(theInterval, null, null);
     DataSourcesResource dataSourcesResource = createResource();
     prepareRequestForAudit();
 
@@ -1189,7 +1192,7 @@ public void testMarkSegmentsAsUnusedInvalidPayload()
         createResource();
 
     final DataSourcesResource.MarkDataSourceSegmentsPayload payload =
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(null, null);
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(null, null, null);
 
     Response response = dataSourcesResource.markSegmentsAsUnused("datasource1", payload, request);
     Assert.assertEquals(400, response.getStatus());
@@ -1203,7 +1206,7 @@ public void testMarkSegmentsAsUnusedInvalidPayloadBothArguments()
         createResource();
 
     final DataSourcesResource.MarkDataSourceSegmentsPayload payload =
-        new DataSourcesResource.MarkDataSourceSegmentsPayload(Intervals.of("2010-01-01/P1D"), ImmutableSet.of());
+        new DataSourcesResource.MarkDataSourceSegmentsPayload(Intervals.of("2010-01-01/P1D"), ImmutableSet.of(), null);
 
     Response response = dataSourcesResource.markSegmentsAsUnused("datasource1", payload, request);
     Assert.assertEquals(400, response.getStatus());