Implement a fingerprinting mechanism to track compaction states in a more efficient manner (#18844)

* meatadata store bits part 1

* annotate segments with compaction fingerprint before persist

* Add ability to generate compaction state fingerprint

* add fingerprint to task context and make legacy last compaction state storage configurable

* update embedded tests for compaction supervisors to flex fingerprints

* checkpoint with persisting compaction states

* add duty to clean up unused compaction states

* take fingerprints into account in CompactionStatus

* Add and improve tests

* get rid of some todo comments

* fix checkstyle

* cleanup some more TODO

* Add some docs

* update web console

* make cache size configurable and fix some spelling

* fixup use of deprecated builder

* fix checktyle

* fix coordinator compactsegments duty and respond to self review comments

* fix spellchecker

* predates is a word

* improve some javadocs

* simplify some test assertions based on review

* better naming

* controller impl cleanup

* For compaction supervisors, take persisting pending compaction states out of hot path

* use Configs.valueOrDefault helper in data segment

* Refactor where fingerprinting happens and how the object mapper is wired up

* refactor CompactionStateManager into an interface with a persisted and heap impl

* remove fingerprinting support from the coordinator compact segments duty

* Move on heap compaction state manager to test sources

* CompactionStateManager is now overlord only

* Refactor how the compaction state fingerprint cache is wired up

* prettify

* small changes after self-review

* Cleanup CompactionStateCache per review

* compactionstatemanager to compactionstatestorage plus refactor

* Add compaction state added and deleted metrics

* improve queries for compaction state cache sync

* clean up doc wording

* Miscl. cleanup from review

* some metadata store code cleanup

* refactor id out of the compaction states table as it is superflous

* Some CompactionStatus cleanup

* Migrate the location of creating a compaction state from config

* More refactoring per review

* refactor to remove duplicate fingerprint generator code

* Do some consolidation of fingerprint related classes to clean up code

* minor cleanup

* fix fobidden api use

* Improvements and cleanup to the fingerprint and state persist + cache

* Refactor where in the code compaction fingerprints are generated

* Formalize unique constraint exception check in sqlmetadataconnector and db specific impls

* some naming cleanup

* Migrate the compaction state cleanup duty to the overlord

* Blow up the compaction supervisor scheduler if incremental caching is disabled

* add some strict input sanitization in upserting compaction fingerprints

* cleanup test class

* Add pending flag to compaction state to prevent potentially destructive early cleanup

* Refactor database naming to use indexingState instead of compactionState

* Refactor naming to IndexingState for the metadata cleanup duty

* refresh some docs

* fixup tests

* Refactoring name of CompactionStateCache to IndexingStateCache

* Rename CompactionStateStorage to IndexingStateStorage

* Refactor compactionStateFingerprint out of the code in favor of indexingStateFingerprint

* Refactor FingerprintMapper name to remove compaction for indexing state

* refactorings after self review

* fixup a few things post merge with master

* Cleanup and refactor after code review round

Batch marking of indexing states as active to avoid chained updates where only one is needed

Build segments table missing columns error column by column

refactor how we are configuring and executing the ol metadata cleanup duties.

fix missed naming refactor

Improve readability of upsertIndexingState

Fixup SqlIndexingStateStorage constructor

drop default impl of isUniqueConstraintViolation

Refactor how the deterministic mapper is handled for reindexing

* cleanup

* use effective state for dimspec and indexspec for reindexing fingerprinting

* Only call into running checks if there are unknown states to check

Author: capistrant

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()
         )
     );