Add a configurable bufferPeriod between when a segment is marked unused and deleted by KillUnusedSegments duty

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

@@ -88,4 +88,9 @@ default void exportTable(
   void createSupervisorsTable();
 
   void deleteAllRecords(String tableName);
+
+  void alterSegmentTableAddLastUsed();
+
+  void updateSegmentTablePopulateLastUsed();
+
 }

docs/configuration/index.md

@@ -814,6 +814,7 @@ These Coordinator static configurations can be defined in the `coordinator/runti
 |`druid.coordinator.kill.period`|How often to send kill tasks to the indexing service. Value must be greater than `druid.coordinator.period.indexingPeriod`. Only applies if kill is turned on.|P1D (1 Day)|
 |`druid.coordinator.kill.durationToRetain`|Only applies if you set `druid.coordinator.kill.on` to `true`. This value is ignored if `druid.coordinator.kill.ignoreDurationToRetain` is `true`. Valid configurations must be a ISO8601 period. Druid will not kill unused segments whose interval end date is beyond `now - durationToRetain`. `durationToRetain` can be a negative ISO8601 period, which would result in `now - durationToRetain` to be in the future.|`P90D`|
 |`druid.coordinator.kill.ignoreDurationToRetain`|A way to override `druid.coordinator.kill.durationToRetain` and tell the coordinator that you do not care about the end date of unused segment intervals when it comes to killing them. If true, the coordinator considers all unused segments as eligible to be killed.|false|
+|`druid.coordinator.kill.bufferPeriod`|The amount of time that a segment must be unused before it is able to be permanently removed from metadata and deep storage. This can serve as a buffer period to prevent data loss if data ends up being needed after being marked unused.|`PT24H`|
 |`druid.coordinator.kill.maxSegments`|Kill at most n unused segments per kill task submission, must be greater than 0. Only applies and MUST be specified if kill is turned on.|100|
 |`druid.coordinator.balancer.strategy`|Specify the type of balancing strategy for the coordinator to use to distribute segments among the historicals. `cachingCost` is logically equivalent to `cost` but is more CPU-efficient on large clusters. `diskNormalized` weights the costs according to the servers' disk usage ratios - there are known issues with this strategy distributing segments unevenly across the cluster. `random` distributes segments among services randomly.|`cost`|
 |`druid.coordinator.balancer.cachingCost.awaitInitialization`|Whether to wait for segment view initialization before creating the `cachingCost` balancing strategy. This property is enabled only when `druid.coordinator.balancer.strategy` is `cachingCost`. If set to 'true', the Coordinator will not start to assign segments, until the segment view is initialized. If set to 'false', the Coordinator will fallback to use the `cost` balancing strategy only if the segment view is not initialized yet. Notes, it may take much time to wait for the initialization since the `cachingCost` balancing strategy involves much computing to build itself.|false|

docs/dependencies/metadata-storage.md

@@ -99,7 +99,9 @@ system. The table has two main functional columns, the other columns are for ind
 Value 1 in the `used` column means that the segment should be "used" by the cluster (i.e., it should be loaded and
 available for requests). Value 0 means that the segment should not be loaded into the cluster. We do this as a means of
 unloading segments from the cluster without actually removing their metadata (which allows for simpler rolling back if
-that is ever an issue).
+that is ever an issue). The `used` column has a corresponding `last_used` column that indicates the date at the instant
+that the `used` status of the segment was last updated. This information can be used by the coordinator to determine if
+a segment is a candidate for deletion (if automated segment killing is enabled).
 
 The `payload` column stores a JSON blob that has all of the metadata for the segment.
 Some of the data in the `payload` column intentionally duplicates data from other columns in the segments table.

docs/operations/upgrade-prep.md

@@ -0,0 +1,114 @@
+---
+id: upgrade-prep
+title: "Upgrade Prep"
+---
+
+<!--
+  ~ 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.
+  -->
+
+## Upgrade to `0.24+` from `0.23` and earlier
+
+### Altering segments table
+
+**If you have set `druid.metadata.storage.connector.createTables` to `true` (which is the default), and your metadata connect user has DDL priviliges, you can disregard this section. You are urged to still evaluate the optional section below**
+
+**The coordinator and overlord services will fail if you do not execute this change prior to the upgrade**
+
+A new column, `last_used`, is needed in the segments table to support new
+segment killing functionality. You can manually alter the table, or you can use
+a pre-written tool to perform the update. 
+
+#### Pre-written tool
+
+Druid provides a `metadata-update` tool for updating Druid's metadata tables.
+
+In the example commands below:
+
+- `lib` is the Druid lib directory
+- `extensions` is the Druid extensions directory
+- `base` corresponds to the value of `druid.metadata.storage.tables.base` in the configuration, `druid` by default.
+- The `--connectURI` parameter corresponds to the value of `druid.metadata.storage.connector.connectURI`.
+- The `--user` parameter corresponds to the value of `druid.metadata.storage.connector.user`.
+- The `--password` parameter corresponds to the value of `druid.metadata.storage.connector.password`.
+- The `--action` parameter corresponds to the update action you are executing. In this case it is: `add-last-used-to-segments`
+
+##### MySQL
+
+```bash
+cd ${DRUID_ROOT}
+java -classpath "lib/*" -Dlog4j.configurationFile=conf/druid/cluster/_common/log4j2.xml -Ddruid.extensions.directory="extensions" -Ddruid.extensions.loadList=[\"mysql-metadata-storage\"] -Ddruid.metadata.storage.type=mysql org.apache.druid.cli.Main tools metadata-update --connectURI="<mysql-uri>" --user <user> --password <pass> --base druid --action add-last-used-to-segments
+```
+
+##### PostgreSQL
+
+```bash
+cd ${DRUID_ROOT}
+java -classpath "lib/*" -Dlog4j.configurationFile=conf/druid/cluster/_common/log4j2.xml -Ddruid.extensions.directory="extensions" -Ddruid.extensions.loadList=[\"postgresql-metadata-storage\"] -Ddruid.metadata.storage.type=postgresql org.apache.druid.cli.Main tools metadata-update --connectURI="<postgresql-uri>" --user <user> --password <pass> --base druid --action add-last-used-to-segments
+```
+
+
+#### Manual ALTER TABLE
+
+```SQL
+ALTER TABLE druid_segments
+ADD last_used varchar(255);
+```
+
+### Populating `last_used` column of the segments table after upgrade (Optional)
+
+This is an optional step to take **after** you upgrade the Overlord and Coordinator to `0.24+` (from `0.23` and earlier). If you do not take this action and are also using `druid.coordinator.kill.on=true`, the logic to identify segments that can be killed will not honor `druid.coordinator.kill.bufferPeriod` for the rows in the segments table where `last_used == null`.
+
+#### Pre-written tool
+
+Druid provides a `metadata-update` tool for updating Druid's metadata tables. Note that this tool will update `last_used` for all rows that match `used = false` in one transaction.
+
+In the example commands below:
+
+- `lib` is the Druid lib directory
+- `extensions` is the Druid extensions directory
+- `base` corresponds to the value of `druid.metadata.storage.tables.base` in the configuration, `druid` by default.
+- The `--connectURI` parameter corresponds to the value of `druid.metadata.storage.connector.connectURI`.
+- The `--user` parameter corresponds to the value of `druid.metadata.storage.connector.user`.
+- The `--password` parameter corresponds to the value of `druid.metadata.storage.connector.password`.
+- The `--action` parameter corresponds to the update action you are executing. In this case it is: `add-last-used-to-segments`
+
+##### MySQL
+
+```bash
+cd ${DRUID_ROOT}
+java -classpath "lib/*" -Dlog4j.configurationFile=conf/druid/cluster/_common/log4j2.xml -Ddruid.extensions.directory="extensions" -Ddruid.extensions.loadList=[\"mysql-metadata-storage\"] -Ddruid.metadata.storage.type=mysql org.apache.druid.cli.Main tools metadata-update --connectURI="<mysql-uri>" --user <user> --password <pass> --base druid --action populate-last-used-column-in-segments
+```
+
+##### PostgreSQL
+
+```bash
+cd ${DRUID_ROOT}
+java -classpath "lib/*" -Dlog4j.configurationFile=conf/druid/cluster/_common/log4j2.xml -Ddruid.extensions.directory="extensions" -Ddruid.extensions.loadList=[\"postgresql-metadata-storage\"] -Ddruid.metadata.storage.type=postgresql org.apache.druid.cli.Main tools metadata-update --connectURI="<postgresql-uri>" --user <user> --password <pass> --base druid --action populate-last-used-column-in-segments
+```
+
+
+#### Manual UPDATE
+
+Note that we choose a random date string for this example. We reccommend using the current UTC time when you invoke the command. If you have lots of rows that match the conditional `used = false`, you may want to incrementlly update the table using a limit clause.
+
+```SQL
+UPDATE druid_segment
+SET last_used = '2022-01-01T00:00:00.000Z'
+where used = false;
+```

server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java

@@ -1006,8 +1006,8 @@ private Set<DataSegment> announceHistoricalSegmentBatch(
 
       PreparedBatch preparedBatch = handle.prepareBatch(
           StringUtils.format(
-              "INSERT INTO %1$s (id, dataSource, created_date, start, %2$send%2$s, partitioned, version, used, payload) "
-                  + "VALUES (:id, :dataSource, :created_date, :start, :end, :partitioned, :version, :used, :payload)",
+              "INSERT INTO %1$s (id, dataSource, created_date, start, %2$send%2$s, partitioned, version, used, payload, last_used) "
+                  + "VALUES (:id, :dataSource, :created_date, :start, :end, :partitioned, :version, :used, :payload, :last_used)",
               dbTables.getSegmentsTable(),
               connector.getQuoteString()
           )
@@ -1024,7 +1024,8 @@ private Set<DataSegment> announceHistoricalSegmentBatch(
               .bind("partitioned", (segment.getShardSpec() instanceof NoneShardSpec) ? false : true)
               .bind("version", segment.getVersion())
               .bind("used", usedSegments.contains(segment))
-              .bind("payload", jsonMapper.writeValueAsBytes(segment));
+              .bind("payload", jsonMapper.writeValueAsBytes(segment))
+              .bind("last_used", DateTimes.nowUtc().toString());
         }
         final int[] affectedRows = preparedBatch.execute();
         final boolean succeeded = Arrays.stream(affectedRows).allMatch(eachAffectedRows -> eachAffectedRows == 1);

server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java

@@ -25,6 +25,7 @@
 import com.google.common.collect.ImmutableList;
 import org.apache.commons.dbcp2.BasicDataSource;
 import org.apache.commons.dbcp2.BasicDataSourceFactory;
+import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.RetryUtils;
 import org.apache.druid.java.util.common.StringUtils;
@@ -44,6 +45,8 @@
 
 import javax.annotation.Nullable;
 import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.SQLRecoverableException;
 import java.sql.SQLTransientException;
@@ -208,6 +211,41 @@ public Void withHandle(Handle handle)
     }
   }
 
+  /**
+   * Execute the desired ALTER statement on the desired table
+   *
+   * @param tableName The name of the table being altered
+   * @param sql ALTER statment to be executed
+   */
+  public void alterTable(final String tableName, final Iterable<String> sql)
+  {
+    try {
+      retryWithHandle(
+          new HandleCallback<Void>()
+          {
+            @Override
+            public Void withHandle(Handle handle)
+            {
+              if (tableExists(handle, tableName)) {
+                log.info("Altering table[%s]", tableName);
+                final Batch batch = handle.createBatch();
+                for (String s : sql) {
+                  batch.add(s);
+                }
+                batch.execute();
+              } else {
+                log.info("Table[%s] doesn't exists", tableName);
+              }
+              return null;
+            }
+          }
+      );
+    }
+    catch (Exception e) {
+      log.warn(e, "Exception Altering table[%s]", tableName);
+    }
+  }
+
   public void createPendingSegmentsTable(final String tableName)
   {
     createTable(
@@ -277,6 +315,7 @@ public void createSegmentTable(final String tableName)
                 + "  version VARCHAR(255) NOT NULL,\n"
                 + "  used BOOLEAN NOT NULL,\n"
                 + "  payload %2$s NOT NULL,\n"
+                + "  last_used VARCHAR(255) NOT NULL,\n"
                 + "  PRIMARY KEY (id)\n"
                 + ")",
                 tableName, getPayloadType(), getQuoteString(), getCollation()
@@ -408,6 +447,50 @@ tableName, getSerialType(), getPayloadType()
     );
   }
 
+  /**
+   * Adds the last_used column to the Druid segment table.
+   *
+   * This is public due to allow the UpdateTables cli tool to use for upgrade prep.
+   */
+  @Override
+  public void alterSegmentTableAddLastUsed()
+  {
+    String tableName = tablesConfigSupplier.get().getSegmentsTable();
+    if (!tableHasColumn(tableName, "last_used")) {
+      log.info("Adding last_used column to %s", tableName);
+      alterTable(
+          tableName,
+          ImmutableList.of(
+              StringUtils.format(
+                  "ALTER TABLE %1$s \n"
+                  + "ADD last_used varchar(255)",
+                  tableName
+              )
+          )
+      );
+    } else {
+      log.info("%s already has last_used column", tableName);
+    }
+  }
+
+  /**
+   * Populates the last_used column for all unused segments in the Druid segment table.
+   *
+   * The current UTC timestamp string is used for the content of each column.
+   * This is public due to allow the UpdateTables cli tool to use in a optional post-upgrade action.
+   */
+  @Override
+  public void updateSegmentTablePopulateLastUsed()
+  {
+    String tableName = tablesConfigSupplier.get().getSegmentsTable();
+    getDBI().withHandle(
+        (Handle handle) -> handle
+            .createStatement(StringUtils.format("UPDATE %s SET last_used = :last_used WHERE used = false", tableName))
+            .bind("last_used", DateTimes.nowUtc().toString())
+            .execute()
+    );
+  }
+
   @Override
   public Void insertOrUpdate(
       final String tableName,
@@ -553,7 +636,11 @@ public void createSegmentTable()
   {
     if (config.get().isCreateTables()) {
       createSegmentTable(tablesConfigSupplier.get().getSegmentsTable());
+      alterSegmentTableAddLastUsed();
     }
+    // Called outside of the above conditional because we want to validate the table
+    // regardless of cluster configuration for creating tables.
+    validateSegmentTable();
   }
 
   @Override
@@ -765,4 +852,52 @@ public Void withHandle(Handle handle)
       log.warn(e, "Exception while deleting records from table");
     }
   }
+
+  /**
+   * Interrogate table metadata and return true or false depending on the existance of the indicated column
+   *
+   * @param tableName The table being interrogated
+   * @param columnName The column being looked for
+   * @return boolean indicating the existence of the column in question
+   */
+  private boolean tableHasColumn(String tableName, String columnName)
+  {
+    return getDBI().withHandle(
+        new HandleCallback<Boolean>()
+        {
+          @Override
+          public Boolean withHandle(Handle handle)
+          {
+            try {
+              DatabaseMetaData dbMetaData = handle.getConnection().getMetaData();
+              ResultSet columns = dbMetaData.getColumns(
+                  null,
+                  null,
+                  tableName,
+                  columnName
+              );
+              return columns.next();
+            }
+            catch (SQLException e) {
+              return false;
+            }
+          }
+        }
+    );
+  }
+
+  /**
+   * Ensure that the segment table has the proper schema required to run Druid properly.
+   *
+   * Throws RuntimeException if the column does not exist. There is no recovering from an invalid schema,
+   * the program should crash.
+   */
+  private void validateSegmentTable()
+  {
+    if (tableHasColumn(tablesConfigSupplier.get().getSegmentsTable(), "last_used")) {
+      return;
+    } else {
+      throw new RuntimeException("Invalid Segment Table Schema! No last_used column!");
+    }
+  }
 }

server/src/main/java/org/apache/druid/metadata/SQLMetadataSegmentPublisher.java

@@ -55,8 +55,8 @@ public SQLMetadataSegmentPublisher(
     this.config = config;
     this.connector = connector;
     this.statement = StringUtils.format(
-        "INSERT INTO %1$s (id, dataSource, created_date, start, %2$send%2$s, partitioned, version, used, payload) "
-        + "VALUES (:id, :dataSource, :created_date, :start, :end, :partitioned, :version, :used, :payload)",
+        "INSERT INTO %1$s (id, dataSource, created_date, start, %2$send%2$s, partitioned, version, used, payload, last_used) "
+        + "VALUES (:id, :dataSource, :created_date, :start, :end, :partitioned, :version, :used, :payload, :last_used)",
         config.getSegmentsTable(), connector.getQuoteString()
     );
   }
@@ -73,7 +73,8 @@ public void publishSegment(final DataSegment segment) throws IOException
         (segment.getShardSpec() instanceof NoneShardSpec) ? false : true,
         segment.getVersion(),
         true,
-        jsonMapper.writeValueAsBytes(segment)
+        jsonMapper.writeValueAsBytes(segment),
+        DateTimes.nowUtc().toString()
     );
   }
 
@@ -87,7 +88,8 @@ void publishSegment(
       final boolean partitioned,
       final String version,
       final boolean used,
-      final byte[] payload
+      final byte[] payload,
+      final String lastUsed
   )
   {
     try {
@@ -128,6 +130,7 @@ public Void withHandle(Handle handle)
                     .bind("version", version)
                     .bind("used", used)
                     .bind("payload", payload)
+                    .bind("last_used", lastUsed)
                     .execute();
 
               return null;