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

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

@@ -22,6 +22,8 @@
 import com.google.common.collect.ImmutableList;
 import org.apache.druid.client.DataSourcesSnapshot;
 import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.segment.metadata.HeapMemoryCompactionStateManager;
+import org.apache.druid.segment.metadata.NoopCompactionStateCache;
 import org.apache.druid.server.compaction.CompactionCandidateSearchPolicy;
 import org.apache.druid.server.compaction.CompactionSegmentIterator;
 import org.apache.druid.server.compaction.NewestSegmentFirstPolicy;
@@ -135,7 +137,9 @@ public void measureNewestSegmentFirstPolicy(Blackhole blackhole)
         policy,
         compactionConfigs,
         dataSources,
-        Collections.emptyMap()
+        Collections.emptyMap(),
+        new HeapMemoryCompactionStateManager(),
+        new NoopCompactionStateCache()
     );
     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`|
+|`legacyPersistLastCompactionStateInSegments`|Whether to persist the full compaction state in segment metadata. When `true` (default), compaction state is stored in both the segment metadata and the compaction 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 compaction 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 compaction state fingerprinting (< Druid 36) 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.compactionStates`|The table to use to store compaction state fingerprints.|`druid_compactionStates`|
 |`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`|

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

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

@@ -31,6 +31,8 @@
 import org.apache.druid.java.util.common.granularity.Granularity;
 import org.apache.druid.query.DruidMetrics;
 import org.apache.druid.rpc.UpdateResponse;
+import org.apache.druid.segment.metadata.CompactionStateManager;
+import org.apache.druid.server.compaction.CompactionStatus;
 import org.apache.druid.server.coordinator.ClusterCompactionConfig;
 import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
 import org.apache.druid.server.coordinator.InlineSchemaDataSourceCompactionConfig;
@@ -98,14 +100,14 @@ public EmbeddedDruidCluster createCluster()
   private void configureCompaction(CompactionEngine compactionEngine)
   {
     final UpdateResponse updateResponse = cluster.callApi().onLeaderOverlord(
-        o -> o.updateClusterCompactionConfig(new ClusterCompactionConfig(1.0, 100, null, true, compactionEngine))
+        o -> o.updateClusterCompactionConfig(new ClusterCompactionConfig(1.0, 100, null, true, compactionEngine, true))
     );
     Assertions.assertTrue(updateResponse.isSuccess());
   }
 
   @MethodSource("getEngine")
   @ParameterizedTest(name = "compactionEngine={0}")
-  public void test_ingestDayGranularity_andCompactToMonthGranularity_withInlineConfig(CompactionEngine compactionEngine)
+  public void test_ingestDayGranularity_andCompactToMonthGranularity_andCompactToYearGranularity_withInlineConfig(CompactionEngine compactionEngine)
   {
     configureCompaction(compactionEngine);
 
@@ -119,7 +121,7 @@ public void test_ingestDayGranularity_andCompactToMonthGranularity_withInlineCon
     Assertions.assertEquals(3, getNumSegmentsWith(Granularities.DAY));
 
     // Create a compaction config with MONTH granularity
-    InlineSchemaDataSourceCompactionConfig compactionConfig =
+    InlineSchemaDataSourceCompactionConfig monthGranularityConfig =
         InlineSchemaDataSourceCompactionConfig
             .builder()
             .forDataSource(dataSource)
@@ -152,11 +154,159 @@ public void test_ingestDayGranularity_andCompactToMonthGranularity_withInlineCon
             )
             .build();
 
-    runCompactionWithSpec(compactionConfig);
+    runCompactionWithSpec(monthGranularityConfig);
     waitForAllCompactionTasksToFinish();
 
     Assertions.assertEquals(0, getNumSegmentsWith(Granularities.DAY));
     Assertions.assertEquals(1, getNumSegmentsWith(Granularities.MONTH));
+
+    verifyCompactedSegmentsHaveFingerprints(monthGranularityConfig);
+
+    InlineSchemaDataSourceCompactionConfig yearGranConfig =
+        InlineSchemaDataSourceCompactionConfig
+            .builder()
+            .forDataSource(dataSource)
+            .withSkipOffsetFromLatest(Period.seconds(0))
+            .withGranularitySpec(
+                new UserCompactionTaskGranularityConfig(Granularities.YEAR, null, null)
+            )
+            .withTuningConfig(
+                new UserCompactionTaskQueryTuningConfig(
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    new DimensionRangePartitionsSpec(null, 5000, List.of("item"), false),
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null
+                )
+            )
+            .build();
+
+    overlord.latchableEmitter().flush(); // flush events so wait for works correctly on the next round of compaction
+    runCompactionWithSpec(yearGranConfig);
+    waitForAllCompactionTasksToFinish();
+
+    Assertions.assertEquals(0, getNumSegmentsWith(Granularities.DAY));
+    Assertions.assertEquals(0, getNumSegmentsWith(Granularities.MONTH));
+    Assertions.assertEquals(1, getNumSegmentsWith(Granularities.YEAR));
+
+    verifyCompactedSegmentsHaveFingerprints(yearGranConfig);
+  }
+
+  @MethodSource("getEngine")
+  @ParameterizedTest(name = "compactionEngine={0}")
+  public void test_compaction_withPersistLastCompactionStateFalse_storesOnlyFingerprint(CompactionEngine compactionEngine)
+      throws InterruptedException
+  {
+    // Configure cluster with persistLastCompactionState=false
+    final UpdateResponse updateResponse = cluster.callApi().onLeaderOverlord(
+        o -> o.updateClusterCompactionConfig(
+            new ClusterCompactionConfig(1.0, 100, null, true, compactionEngine, false)
+        )
+    );
+    Assertions.assertTrue(updateResponse.isSuccess());
+
+    // Ingest data at DAY granularity
+    runIngestionAtGranularity(
+        "DAY",
+        "2025-06-01T00:00:00.000Z,shirt,105\n"
+        + "2025-06-02T00:00:00.000Z,trousers,210"
+    );
+    Assertions.assertEquals(2, getNumSegmentsWith(Granularities.DAY));
+
+    // Create compaction config to compact to MONTH granularity
+    InlineSchemaDataSourceCompactionConfig monthConfig =
+        InlineSchemaDataSourceCompactionConfig
+            .builder()
+            .forDataSource(dataSource)
+            .withSkipOffsetFromLatest(Period.seconds(0))
+            .withGranularitySpec(
+                new UserCompactionTaskGranularityConfig(Granularities.MONTH, null, null)
+            )
+            .withTuningConfig(
+                new UserCompactionTaskQueryTuningConfig(
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    new DimensionRangePartitionsSpec(1000, null, List.of("item"), false),
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null
+                )
+            )
+            .build();
+
+    runCompactionWithSpec(monthConfig);
+    waitForAllCompactionTasksToFinish();
+
+    verifySegmentsHaveNullLastCompactionStateAndNonNullFingerprint();
+  }
+
+  private void verifySegmentsHaveNullLastCompactionStateAndNonNullFingerprint()
+  {
+    overlord
+        .bindings()
+        .segmentsMetadataStorage()
+        .retrieveAllUsedSegments(dataSource, Segments.ONLY_VISIBLE)
+        .forEach(segment -> {
+          Assertions.assertNull(
+              segment.getLastCompactionState(),
+              "Segment " + segment.getId() + " should have null lastCompactionState"
+          );
+          Assertions.assertNotNull(
+              segment.getCompactionStateFingerprint(),
+              "Segment " + segment.getId() + " should have non-null compactionStateFingerprint"
+          );
+        });
+  }
+
+  private void verifyCompactedSegmentsHaveFingerprints(DataSourceCompactionConfig compactionConfig)
+  {
+    CompactionStateManager compactionStateManager = overlord
+        .bindings()
+        .getInstance(CompactionStateManager.class);
+
+    String expectedFingerprint = compactionStateManager.generateCompactionStateFingerprint(
+        CompactionStatus.createCompactionStateFromConfig(compactionConfig),
+        dataSource
+    );
+
+    overlord
+        .bindings()
+        .segmentsMetadataStorage()
+        .retrieveAllUsedSegments(dataSource, Segments.ONLY_VISIBLE)
+        .forEach(segment -> {
+          Assertions.assertEquals(
+              expectedFingerprint,
+              segment.getCompactionStateFingerprint(),
+              "Segment " + segment.getId() + " fingerprint should match expected fingerprint"
+          );
+        });
   }
 
   private void runCompactionWithSpec(DataSourceCompactionConfig config)

embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaClusterMetricsTest.java

@@ -212,7 +212,7 @@ public void test_ingestClusterMetrics_withConcurrentCompactionSupervisor_andSkip
     );
 
     final ClusterCompactionConfig updatedCompactionConfig
-        = new ClusterCompactionConfig(1.0, 10, null, true, null);
+        = new ClusterCompactionConfig(1.0, 10, null, true, null, null);
     final UpdateResponse updateResponse = cluster.callApi().onLeaderOverlord(
         o -> o.updateClusterCompactionConfig(updatedCompactionConfig)
     );
@@ -323,7 +323,7 @@ public void test_ingestClusterMetrics_compactionSkipsLockedIntervals()
     );
 
     final ClusterCompactionConfig updatedCompactionConfig
-        = new ClusterCompactionConfig(1.0, 10, null, true, null);
+        = new ClusterCompactionConfig(1.0, 10, null, true, null, null);
     final UpdateResponse updateResponse = cluster.callApi().onLeaderOverlord(
         o -> o.updateClusterCompactionConfig(updatedCompactionConfig)
     );

extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/compact/CatalogCompactionTest.java

@@ -164,7 +164,7 @@ private IndexTask createIndexTaskForInlineData(String taskId)
   private void enableCompactionSupervisor()
   {
     final UpdateResponse updateResponse = cluster.callApi().onLeaderOverlord(
-        o -> o.updateClusterCompactionConfig(new ClusterCompactionConfig(1.0, 10, null, true, null))
+        o -> o.updateClusterCompactionConfig(new ClusterCompactionConfig(1.0, 10, null, true, null, null))
     );
     Assertions.assertTrue(updateResponse.isSuccess());
   }

indexing-hadoop/src/main/java/org/apache/druid/indexer/updater/MetadataStorageUpdaterJobSpec.java

@@ -99,6 +99,7 @@ public MetadataStorageTablesConfig getMetadataStorageTablesConfig()
         null,
         null,
         null,
+        null,
         null
     );
   }

indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java

@@ -641,6 +641,25 @@ public static boolean isGuaranteedRollup(
     return tuningConfig.isForceGuaranteedRollup();
   }
 
+  /**
+   * Returns a function that adds the given compaction state fingerprint to all segments.
+   * If the fingerprint is null, returns an identity function that leaves segments unchanged.
+   */
+  public static Function<Set<DataSegment>, Set<DataSegment>> addCompactionStateFingerprintToSegments(
+      String compactionStateFingerprint
+  )
+  {
+    if (compactionStateFingerprint != null) {
+      return segments -> segments.stream()
+                                 .map(
+                                     segment -> segment.withCompactionStateFingerprint(compactionStateFingerprint)
+                                 )
+                                 .collect(Collectors.toSet());
+    } else {
+      return Function.identity();
+    }
+  }
+
   public static Function<Set<DataSegment>, Set<DataSegment>> addCompactionStateToSegments(
       boolean storeCompactionState,
       TaskToolbox toolbox,

indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java

@@ -902,11 +902,19 @@ private TaskStatus generateAndPublishSegments(
           Tasks.STORE_COMPACTION_STATE_KEY,
           Tasks.DEFAULT_STORE_COMPACTION_STATE
       );
+
+      final String compactionStateFingerprint = getContextValue(
+          Tasks.COMPACTION_STATE_FINGERPRINT_KEY,
+          null
+      );
+
       final Function<Set<DataSegment>, Set<DataSegment>> annotateFunction =
           addCompactionStateToSegments(
               storeCompactionState,
               toolbox,
               ingestionSchema
+          ).andThen(
+              addCompactionStateFingerprintToSegments(compactionStateFingerprint)
           );
 
       Set<DataSegment> tombStones = Collections.emptySet();

indexing-service/src/main/java/org/apache/druid/indexing/common/task/Tasks.java

@@ -68,4 +68,13 @@ public class Tasks
   static {
     Verify.verify(STORE_COMPACTION_STATE_KEY.equals(CompactSegments.STORE_COMPACTION_STATE_KEY));
   }
+
+  /**
+   * Context k:v pair that holds the fingerprint of the compaction state to be stored with the segment
+   */
+  public static final String COMPACTION_STATE_FINGERPRINT_KEY = "compactionStateFingerprint";
+
+  static {
+    Verify.verify(COMPACTION_STATE_FINGERPRINT_KEY.equals(CompactSegments.COMPACTION_STATE_FINGERPRINT_KEY));
+  }
 }

indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java

@@ -1162,12 +1162,20 @@ private void publishSegments(
         Tasks.STORE_COMPACTION_STATE_KEY,
         Tasks.DEFAULT_STORE_COMPACTION_STATE
     );
-    final Function<Set<DataSegment>, Set<DataSegment>> annotateFunction = addCompactionStateToSegments(
-        storeCompactionState,
-        toolbox,
-        ingestionSchema
+    final String compactionStateFingerprint = getContextValue(
+        Tasks.COMPACTION_STATE_FINGERPRINT_KEY,
+        null
     );
 
+    final Function<Set<DataSegment>, Set<DataSegment>> annotateFunction =
+        addCompactionStateToSegments(
+            storeCompactionState,
+            toolbox,
+            ingestionSchema
+        ).andThen(
+            addCompactionStateFingerprintToSegments(compactionStateFingerprint)
+        );
+
     Set<DataSegment> tombStones = Collections.emptySet();
     if (getIngestionMode() == IngestionMode.REPLACE) {
       TombstoneHelper tombstoneHelper = new TombstoneHelper(toolbox.getTaskActionClient());