Implement a fingerprinting mechanism to track compaction states in a more efficient manner

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

@@ -21,7 +21,10 @@
 
 import com.google.common.collect.ImmutableList;
 import org.apache.druid.client.DataSourcesSnapshot;
+import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.segment.metadata.DefaultIndexingStateFingerprintMapper;
+import org.apache.druid.segment.metadata.NoopIndexingStateCache;
 import org.apache.druid.server.compaction.CompactionCandidateSearchPolicy;
 import org.apache.druid.server.compaction.CompactionSegmentIterator;
 import org.apache.druid.server.compaction.NewestSegmentFirstPolicy;
@@ -135,7 +138,8 @@ public void measureNewestSegmentFirstPolicy(Blackhole blackhole)
         policy,
         compactionConfigs,
         dataSources,
-        Collections.emptyMap()
+        Collections.emptyMap(),
+        new DefaultIndexingStateFingerprintMapper(new NoopIndexingStateCache(), new DefaultObjectMapper())
     );
     for (int i = 0; i < numCompactionTaskSlots && iterator.hasNext(); i++) {
       blackhole.consume(iterator.next());

docs/api-reference/automatic-compaction-api.md

@@ -889,6 +889,7 @@ This includes the following fields:
 |`compactionPolicy`|Policy to choose intervals for compaction. Currently, the only supported policy is [Newest segment first](#compaction-policy-newestsegmentfirst).|Newest segment first|
 |`useSupervisors`|Whether compaction should be run on Overlord using supervisors instead of Coordinator duties.|false|
 |`engine`|Engine used for running compaction tasks, unless overridden in the datasource-level compaction config. Possible values are `native` and `msq`. `msq` engine can be used for compaction only if `useSupervisors` is `true`.|`native`|
+|`storeCompactionStatePerSegment`|**This configuration only takes effect if `useSupervisors` is `true`.** Whether to persist the full compaction state in segment metadata. When `true` (default), compaction state is stored in both the segment metadata and the indexing states table. This is historically how Druid has worked. When `false`, only a fingerprint reference is stored in the segment metadata, reducing storage overhead in the segments table. The actual compaction state is stored in the indexing states table and can be referenced with the aforementioned fingerprint. Eventually this configuration will be removed and all compaction will use the fingerprint method only. This configuration exists for operators to opt into this future pattern early. **WARNING: if you set this to false and then compact data, rolling back to a Druid version that predates indexing state fingerprinting (< Druid 37) will result in missing compaction states and trigger compaction on segments that may already be compacted.**|`true`|
 
 #### Compaction policy `newestSegmentFirst`
 

docs/configuration/index.md

@@ -389,6 +389,7 @@ These properties specify the JDBC connection and other configuration around the
 |`druid.metadata.storage.tables.segments`|The table to use to look for segments.|`druid_segments`|
 |`druid.metadata.storage.tables.rules`|The table to use to look for segment load/drop rules.|`druid_rules`|
 |`druid.metadata.storage.tables.config`|The table to use to look for configs.|`druid_config`|
+|`druid.metadata.storage.tables.indexingStates`|The table that stores indexing state payloads and fingerprints.|`druid_indexingStates`|
 |`druid.metadata.storage.tables.tasks`|Used by the indexing service to store tasks.|`druid_tasks`|
 |`druid.metadata.storage.tables.taskLog`|Used by the indexing service to store task logs.|`druid_tasklogs`|
 |`druid.metadata.storage.tables.taskLock`|Used by the indexing service to store task locks.|`druid_tasklocks`|

docs/data-management/automatic-compaction.md

@@ -241,12 +241,18 @@ You can run automatic compaction using compaction supervisors on the Overlord ra
 * Can use either the native compaction engine or the [MSQ task engine](#use-msq-for-auto-compaction)
 * More reactive and submits tasks as soon as a compaction slot is available
 * Tracked compaction task status to avoid re-compacting an interval repeatedly
+* Uses new Indexing State Fingerprinting mechanisms to store less data per segment in metadata storage
 
 
-To use compaction supervisors, update the [compaction dynamic config](../api-reference/automatic-compaction-api.md#update-cluster-level-compaction-config) and set:
+To use compaction supervisors, the following configuration requirements must be met:
 
-*  `useSupervisors` to `true` so that compaction tasks can be run as supervisor tasks
-*  `engine` to `msq` to use the MSQ task engine as the compaction engine or to `native` (default value) to use the native engine.
+* You must be using incremental segment metadata caching:
+  * `druid.manager.segments.useIncrementalCache` set to `always` or `ifSynced` in your Overlord and Coordinator runtime properties.
+    * See [Segment metadata caching](../configuration/index.md#metadata-retrieval) for full configuration documentation.
+
+* update the [compaction dynamic config](../api-reference/automatic-compaction-api.md#update-cluster-level-compaction-config) and set:
+  *  `useSupervisors` to `true` so that compaction tasks can be run as supervisor tasks
+  *  `engine` to `msq` to use the MSQ task engine as the compaction engine or to `native` (default value) to use the native engine.
 
 Compaction supervisors use the same syntax as auto-compaction using Coordinator duties with one key difference: you submit the auto-compaction as a supervisor spec. In the spec, set the `type` to `autocompact` and include the auto-compaction config in the `spec`.
 

docs/operations/clean-metadata-store.md

@@ -34,6 +34,7 @@ The metadata store includes the following:
 - Compaction configuration records
 - Datasource records created by supervisors
 - Indexer task logs
+- Indexing State records
 
 When you delete some entities from Apache Druid, records related to the entity may remain in the metadata store.
 If you have a high datasource churn rate, meaning you frequently create and delete many short-lived datasources or other related entities like compaction configuration or rules, the leftover records can fill your metadata store and cause performance issues.
@@ -59,7 +60,7 @@ If you have compliance requirements to keep audit records and you enable automat
 ## Configure automated metadata cleanup
 
 You can configure cleanup for each entity separately, as described in this section.
-Define the properties in the `coordinator/runtime.properties` file.
+Unless otherwise specified, define the properties in the `coordinator/runtime.properties` file.
 
 The cleanup of one entity may depend on the cleanup of another entity as follows:
 - You have to configure a [kill task for segment records](#segment-records-and-segments-in-deep-storage-kill-task) before you can configure automated cleanup for [rules](#rules-records) or [compaction configuration](#compaction-configuration-records).
@@ -131,6 +132,27 @@ Compaction configuration cleanup uses the following configuration:
 If you already have an extremely large compaction configuration, you may not be able to delete compaction configuration due to size limits with the audit log. In this case you can set `druid.audit.manager.maxPayloadSizeBytes` and `druid.audit.manager.skipNullField` to avoid the auditing issue. See [Audit logging](../configuration/index.md#audit-logging).
 :::
 
+### Indexing State Records
+
+:::info
+Indexing State Records are cleaned up by the overlord. Therefore, this section should be configured in the `overlord/runtime.properties` file.
+:::
+
+:::info
+Indexing State Records are only created if you are using automatic compaction supervisors.
+:::
+
+Indexing State records become eligible for deletion in the following scenarios:
+- When no `used` segments have an `indexing_state_fingerprint` that is equal to the `fingerprint` of the record.
+- When a record has `pending` state set to `true`
+
+Indexing State cleanup uses the following configuration:
+ - `druid.overlord.kill.indexingStates.on`: When `true`, enables cleanup for indexing state records.
+ - `druid.overlord.kill.indexingStates.period`: Defines the frequency in [ISO 8601 format](https://en.wikipedia.org/wiki/ISO_8601#Durations) for the cleanup job to check for and delete eligible indexing state records. Defaults to `P1D`.
+ - `druid.overlord.kill.indexingStates.durationToRetain`:  Defines the retention period in [ISO 8601 format](https://en.wikipedia.org/wiki/ISO_8601#Durations) after indexing state records are marked as `used=false` become eligible for deletion. Defaults to `P7D`.
+ - `druid.overlord.kill.indexingStates.pendingDurationToRetain`: Defines the retention period in [ISO 8601 format](https://en.wikipedia.org/wiki/ISO_8601#Durations) after creation that pending indexing state records become eligible for deletion. Defaults to `P7D`.
+   - It is recommended that this value be greater than the maximum expected duration of compaction tasks to avoid pending records being deleted prematurely.
+
 ### Datasource records created by supervisors
 
 Datasource records created by supervisors become eligible for deletion when the supervisor is terminated or does not exist in the `druid_supervisors` table and the `durationToRetain` time has passed since their creation.
@@ -160,7 +182,9 @@ For more detail, see [Task logging](../configuration/index.md#task-logging).
 ## Disable automated metadata cleanup
 
 Druid automatically cleans up metadata records, excluding compaction configuration records and indexer task logs.
-To disable automated metadata cleanup, set the following properties in the `coordinator/runtime.properties` file:
+To disable automated metadata cleanup
+
+set the following properties in the `coordinator/runtime.properties` file:
 
 ```properties
 # Keep unused segments
@@ -178,11 +202,19 @@ druid.coordinator.kill.rule.on=false
 # Keep datasource records created by supervisors
 druid.coordinator.kill.datasource.on=false
 ```
+set the following properties in the `overlord/runtime.properties` file:
+
+```properties
+# Keep indexing state records
+druid.overlord.kill.indexingStates.on=false
+```
 
 ## Example configuration for automated metadata cleanup
 
 Consider a scenario where you have scripts to create and delete hundreds of datasources and related entities a day. You do not want to fill your metadata store with leftover records. The datasources and related entities tend to persist for only one or two days. Therefore, you want to run a cleanup job that identifies and removes leftover records that are at least four days old after a seven day buffer period in case you want to recover the data. The exception is for audit logs, which you need to retain for 30 days:
 
+Coordinator configuration (`coordinator/runtime.properties`):
+
 ```properties
 ...
 # Schedule the metadata management store task for every hour:
@@ -226,6 +258,18 @@ druid.coordinator.kill.datasource.durationToRetain=P4D
 ...
 ```
 
+Overlord configuration - if using automatic compaction supervisors (`overlord/runtime.properties`):
+
+```properties
+...
+# Poll every day to delete pending or unreferenced indexing state records > 4 days old
+druid.overlord.kill.indexingStates.on=true
+druid.overlord.kill.indexingStates.period=P1D
+druid.overlord.kill.indexingStates.durationToRetain=P4D
+druid.overlord.kill.indexingStates.pendingDurationToRetain=P4D
+...
+```
+
 ## Learn more
 See the following topics for more information:
 - [Metadata management](../configuration/index.md#metadata-management) for metadata store configuration reference.

embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/AutoCompactionTest.java

@@ -1508,7 +1508,7 @@ public void testAutoCompactionDutyWithDimensionsSpec(CompactionEngine engine) th
   @ParameterizedTest(name = "useSupervisors={0}")
   public void testAutoCompactionDutyWithFilter(boolean useSupervisors) throws Exception
   {
-    updateClusterConfig(new ClusterCompactionConfig(0.5, 10, null, useSupervisors, null));
+    updateClusterConfig(new ClusterCompactionConfig(0.5, 10, null, useSupervisors, null, true));
 
     loadData(INDEX_TASK);
     try (final Closeable ignored = unloader(fullDatasourceName)) {
@@ -1552,7 +1552,7 @@ public void testAutoCompactionDutyWithFilter(boolean useSupervisors) throws Exce
   @ParameterizedTest(name = "useSupervisors={0}")
   public void testAutoCompationDutyWithMetricsSpec(boolean useSupervisors) throws Exception
   {
-    updateClusterConfig(new ClusterCompactionConfig(0.5, 10, null, useSupervisors, null));
+    updateClusterConfig(new ClusterCompactionConfig(0.5, 10, null, useSupervisors, null, true));
 
     loadData(INDEX_TASK);
     try (final Closeable ignored = unloader(fullDatasourceName)) {
@@ -1854,7 +1854,7 @@ private void forceTriggerAutoCompaction(
           ).collect(Collectors.toList())
       );
       updateClusterConfig(
-          new ClusterCompactionConfig(0.5, intervals.size(), policy, true, null)
+          new ClusterCompactionConfig(0.5, intervals.size(), policy, true, null, true)
       );
 
       // Wait for scheduler to pick up the compaction job
@@ -1864,7 +1864,7 @@ private void forceTriggerAutoCompaction(
 
       // Disable all compaction
       updateClusterConfig(
-          new ClusterCompactionConfig(0.5, intervals.size(), COMPACT_NOTHING_POLICY, true, null)
+          new ClusterCompactionConfig(0.5, intervals.size(), COMPACT_NOTHING_POLICY, true, null, true)
       );
     } else {
       forceTriggerAutoCompaction(numExpectedSegmentsAfterCompaction);
@@ -1956,7 +1956,8 @@ private void updateCompactionTaskSlot(double compactionTaskSlotRatio, int maxCom
             maxCompactionTaskSlots,
             oldConfig.getCompactionPolicy(),
             oldConfig.isUseSupervisors(),
-            oldConfig.getEngine()
+            oldConfig.getEngine(),
+            oldConfig.isStoreCompactionStatePerSegment()
         )
     );