From 180b1e18dc280874979fcc7163f1a969a1c69390 Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Fri, 26 Jun 2026 13:05:38 -0500 Subject: [PATCH 1/4] feat: support mandatory compaction trigger for cascading reindexing supervisors who have unapplied deletion rules --- .../api-reference/automatic-compaction-api.md | 1 + docs/data-management/cascading-reindexing.md | 6 + .../compact/CompactionSupervisorTest.java | 53 +++++++ .../CompactionConfigBasedJobTemplate.java | 18 ++- .../compact/ReindexingConfigOptimizer.java | 14 ++ .../ReindexingDeletionRuleOptimizer.java | 25 +++ .../ReindexingDeletionRuleOptimizerTest.java | 142 ++++++++++++++++++ .../CompactionCandidateSearchPolicy.java | 27 ++++ .../MostFragmentedIntervalFirstPolicy.java | 73 ++++++++- ...MostFragmentedIntervalFirstPolicyTest.java | 101 +++++++++++++ 10 files changed, 453 insertions(+), 7 deletions(-) diff --git a/docs/api-reference/automatic-compaction-api.md b/docs/api-reference/automatic-compaction-api.md index 4f2ce3d62afd..613533eb51f7 100644 --- a/docs/api-reference/automatic-compaction-api.md +++ b/docs/api-reference/automatic-compaction-api.md @@ -911,6 +911,7 @@ This experimental policy prioritizes compaction of intervals with the largest nu |`maxAverageUncompactedBytesPerSegment`|Maximum average size of uncompacted segments in an interval eligible for compaction. Human-readable byte format (e.g., "2GiB").|2 GiB| |`minUncompactedBytesPercentForFullCompaction`|Threshold percentage (0-100) of uncompacted bytes to total bytes below which minor compaction is eligible instead of full compaction.|0| |`minUncompactedRowsPercentForFullCompaction`|Threshold percentage (0-100) of uncompacted rows to total rows below which minor compaction is eligible instead of full compaction.|0| +|`forcePendingDeletionCompaction`|When `true`, an interval whose [cascading reindexing](../data-management/cascading-reindexing.md) deletion rules have not yet been applied to all of its segments is compacted even if it does not meet the minimum interval-size thresholds above. This applies only to cascading reindexing supervisors and their deletion rules; it has no effect on intervals without pending deletion rules or on any other compaction. Intended for operators who apply deletion rules for data compliance.|false| #### Compaction policy `fixedIntervalOrder` diff --git a/docs/data-management/cascading-reindexing.md b/docs/data-management/cascading-reindexing.md index 1cc11764ad05..550ecebc3190 100644 --- a/docs/data-management/cascading-reindexing.md +++ b/docs/data-management/cascading-reindexing.md @@ -272,6 +272,12 @@ Example: } ``` +**Applying deletions regardless of compaction thresholds:** + +When the cluster compaction policy is [`mostFragmentedFirst`](../api-reference/automatic-compaction-api.md#compaction-policy-mostfragmentedfirst), an interval must meet minimum size thresholds (such as `minUncompactedCount`) before it is compacted. An interval that recently crossed a deletion rule's `olderThan` threshold may have too few uncompacted segments to qualify, which delays applying the deletion. + +Operators who apply deletion rules for data compliance can set `forcePendingDeletionCompaction: true` on the policy. Any interval that has deletion rules not yet applied to all of its segments is then compacted regardless of those thresholds, while still using the policy's full-vs-minor decision. This applies only to cascading reindexing deletion rules; it does not change eligibility for any other compaction. The setting defaults to `false`, so by default deletions wait until the interval qualifies for compaction on its own. + #### Index spec rules Index spec rules control compression and encoding settings for compacted segments, independently of partitioning. diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java index 34ea243bbbc7..b6cca782795d 100644 --- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java @@ -443,6 +443,59 @@ public void test_cascadingCompactionTemplate_multiplePeriodsApplyDifferentCompac verifyEventCountOlderThan(Period.days(7), "item", "hat", 0); } + @Test + public void test_cascadingReindexing_appliesPendingDeletionRule_evenWhenIntervalBelowFragmentationThreshold() + { + // minUncompactedCount is far above the segment count, so the interval only becomes a candidate + // because forcePendingDeletionCompaction is enabled and a deletion rule is pending. + configureCompaction( + CompactionEngine.MSQ, + new MostFragmentedIntervalFirstPolicy(10_000, null, null, null, null, null, true) + ); + + DateTime now = DateTimes.nowUtc(); + // A few segments older than the rule period; half the rows are item='hat'. + String oldEvents = generateEventsInInterval( + new Interval(now.minusDays(31), now.minusDays(14)), + 6, + Duration.ofHours(25).toMillis() + ); + runIngestionAtGranularity("DAY", oldEvents); + cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator, broker); + + Assertions.assertEquals(6, getNumSegmentsWith(Granularities.DAY)); + verifyEventCountOlderThan(Period.days(7), "item", "hat", 3); + + ReindexingDeletionRule deletionRule = new ReindexingDeletionRule( + "deletionRule", + "Drop rows where item is 'hat'", + Period.days(7), + new EqualityFilter("item", ColumnType.STRING, "hat", null), + null + ); + + CascadingReindexingTemplate template = new CascadingReindexingTemplate( + dataSource, + null, + null, + InlineReindexingRuleProvider.builder().deletionRules(List.of(deletionRule)).build(), + null, + null, + null, + Granularities.DAY, + new DynamicPartitionsSpec(null, null), + null, + null + ); + + runCompactionWithSpec(template); + waitForAllCompactionTasksToFinish(); + cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator, broker); + + // The pending deletion rule forced compaction despite the interval being below the size threshold. + verifyEventCountOlderThan(Period.days(7), "item", "hat", 0); + } + @Test public void test_cascadingReindexing_withVirtualColumnOnNestedData_filtersCorrectly() { diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java index 1a8dd95179b5..0a589b7a90e4 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java @@ -28,7 +28,9 @@ import org.apache.druid.java.util.common.Intervals; import org.apache.druid.java.util.common.granularity.Granularity; import org.apache.druid.server.compaction.CompactionCandidate; +import org.apache.druid.server.compaction.CompactionCandidateSearchPolicy; import org.apache.druid.server.compaction.CompactionSlotManager; +import org.apache.druid.server.compaction.CompactionTaskStatus; import org.apache.druid.server.compaction.DataSourceCompactibleSegmentIterator; import org.apache.druid.server.compaction.Eligibility; import org.apache.druid.server.compaction.NewestSegmentFirstPolicy; @@ -94,10 +96,18 @@ public List createCompactionJobs( // Create a job for each CompactionCandidate while (segmentIterator.hasNext()) { final CompactionCandidate candidate = segmentIterator.next(); - final Eligibility eligibility = - params.getClusterCompactionConfig() - .getCompactionPolicy() - .checkEligibilityForCompaction(candidate, params.getLatestTaskStatus(candidate)); + final CompactionCandidateSearchPolicy policy = params.getClusterCompactionConfig().getCompactionPolicy(); + final CompactionTaskStatus latestTaskStatus = params.getLatestTaskStatus(candidate); + + Eligibility eligibility = policy.checkEligibilityForCompaction(candidate, latestTaskStatus); + if (policy.isForcePendingDeletionCompaction() + && !eligibility.isEligible() + && configOptimizer.hasUnappliedDeletionRules(config, candidate, params)) { + // When the operator has opted in, a cascading reindexing interval with unapplied deletion rules + // must be reindexed for compliance even when it is below the policy's size thresholds; bypass + // those gates but keep its full-vs-minor decision. + eligibility = policy.checkEligibilityForMandatoryCompaction(candidate, latestTaskStatus); + } if (!eligibility.isEligible()) { continue; } diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingConfigOptimizer.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingConfigOptimizer.java index d091d5433f22..d76a46c2dfb0 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingConfigOptimizer.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingConfigOptimizer.java @@ -47,6 +47,20 @@ DataSourceCompactionConfig optimizeConfig( CompactionJobParams params ); + /** + * Whether {@code candidate} has deletion rules not yet applied to all of its segments. The job + * template uses this to force compaction eligibility for intervals that must be reindexed for data + * compliance, regardless of the policy's minimum interval-size thresholds. Defaults to false. + */ + default boolean hasUnappliedDeletionRules( + DataSourceCompactionConfig config, + CompactionCandidate candidate, + CompactionJobParams params + ) + { + return false; + } + /** * Identity finalizer that returns the config unchanged. * Use this for templates that don't need per-candidate customization. diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizer.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizer.java index 0e16ff20a05b..7b96cf697a11 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizer.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizer.java @@ -95,6 +95,31 @@ public DataSourceCompactionConfig optimizeConfig( .build(); } + @Override + public boolean hasUnappliedDeletionRules( + DataSourceCompactionConfig config, + CompactionCandidate candidate, + CompactionJobParams params + ) + { + if (config.getTransformSpec() == null) { + return false; + } + final DimFilter filter = config.getTransformSpec().getFilter(); + if (!(filter instanceof NotDimFilter)) { + return false; + } + + // Compare the decomposed NOT(OR(...)) clauses (as optimizeConfig does) rather than the whole + // transformSpec, so a spec differing only by folded-in partitioning virtual columns isn't mistaken + // for a pending deletion. Unlike optimizeConfig, never-compacted candidates are not short-circuited. + return computeRequiredSetOfFilterRulesForCandidate( + candidate, + (NotDimFilter) filter, + params.getFingerprintMapper() + ) != null; + } + /** * Computes the required set of deletion rules to be applied for the given {@link CompactionCandidate}. *

diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizerTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizerTest.java index 100349d96409..dce3b5ab1c37 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizerTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizerTest.java @@ -629,6 +629,148 @@ public void testOptimize_NoTransformSpec_NoOptimization() Assertions.assertSame(config, result); } + @Test + public void testHasUnappliedDeletionRules_NoFingerprints_ReturnsTrue() + { + DimFilter filterA = new SelectorDimFilter("country", "US", null); + DimFilter filterB = new SelectorDimFilter("country", "UK", null); + NotDimFilter expectedFilter = new NotDimFilter(new OrDimFilter(Arrays.asList(filterA, filterB))); + + CompactionCandidate candidate = createCandidateWithNullFingerprints(2); + InlineSchemaDataSourceCompactionConfig config = createConfigWithFilter(expectedFilter, null); + CompactionJobParams params = createParams(); + + Assertions.assertTrue(optimizer.hasUnappliedDeletionRules(config, candidate, params)); + } + + @Test + public void testHasUnappliedDeletionRules_PartiallyApplied_ReturnsTrue() + { + DimFilter filterA = new SelectorDimFilter("country", "US", null); + DimFilter filterB = new SelectorDimFilter("country", "UK", null); + + CompactionState state = createStateWithSingleFilter(filterA); + indexingStateStorage.upsertIndexingState(TestDataSource.WIKI, "fp1", state, DateTimes.nowUtc()); + syncCacheFromManager(); + + NotDimFilter expectedFilter = new NotDimFilter(new OrDimFilter(Arrays.asList(filterA, filterB))); + CompactionCandidate candidate = createCandidateWithFingerprints("fp1"); + InlineSchemaDataSourceCompactionConfig config = createConfigWithFilter(expectedFilter, null); + CompactionJobParams params = createParams(); + + Assertions.assertTrue(optimizer.hasUnappliedDeletionRules(config, candidate, params)); + } + + @Test + public void testHasUnappliedDeletionRules_AllApplied_ReturnsFalse() + { + DimFilter filterA = new SelectorDimFilter("country", "US", null); + DimFilter filterB = new SelectorDimFilter("country", "UK", null); + + CompactionState state = createStateWithFilters(filterA, filterB); + indexingStateStorage.upsertIndexingState(TestDataSource.WIKI, "fp1", state, DateTimes.nowUtc()); + syncCacheFromManager(); + + NotDimFilter expectedFilter = new NotDimFilter(new OrDimFilter(Arrays.asList(filterA, filterB))); + CompactionCandidate candidate = createCandidateWithFingerprints("fp1"); + InlineSchemaDataSourceCompactionConfig config = createConfigWithFilter(expectedFilter, null); + CompactionJobParams params = createParams(); + + Assertions.assertFalse(optimizer.hasUnappliedDeletionRules(config, candidate, params)); + } + + @Test + public void testHasUnappliedDeletionRules_MissingState_ReturnsTrue() + { + DimFilter filterA = new SelectorDimFilter("country", "US", null); + NotDimFilter expectedFilter = new NotDimFilter(filterA); + + // No cached state for fp1, so the rule can't be proven applied. + CompactionCandidate candidate = createCandidateWithFingerprints("fp1"); + InlineSchemaDataSourceCompactionConfig config = createConfigWithFilter(expectedFilter, null); + CompactionJobParams params = createParams(); + + Assertions.assertTrue(optimizer.hasUnappliedDeletionRules(config, candidate, params)); + } + + @Test + public void testHasUnappliedDeletionRules_NeverCompactedCandidate_ReturnsTrue() + { + DimFilter filterA = new SelectorDimFilter("country", "US", null); + NotDimFilter expectedFilter = new NotDimFilter(filterA); + + // optimizeConfig short-circuits never-compacted candidates; this must still report them unapplied. + List segments = new ArrayList<>(); + segments.add(DataSegment.builder(WIKI_SEGMENT).indexingStateFingerprint(null).build()); + CompactionCandidate candidate = CompactionCandidate.from( + segments, + null, + CompactionStatus.pending(CompactionStatus.NEVER_COMPACTED_REASON) + ); + + InlineSchemaDataSourceCompactionConfig config = createConfigWithFilter(expectedFilter, null); + CompactionJobParams params = createParams(); + + Assertions.assertTrue(optimizer.hasUnappliedDeletionRules(config, candidate, params)); + } + + @Test + public void testHasUnappliedDeletionRules_NoTransformSpec_ReturnsFalse() + { + InlineSchemaDataSourceCompactionConfig config = InlineSchemaDataSourceCompactionConfig.builder() + .forDataSource(TestDataSource.WIKI) + .build(); + CompactionCandidate candidate = createCandidateWithFingerprints("fp1"); + CompactionJobParams params = createParams(); + + Assertions.assertFalse(optimizer.hasUnappliedDeletionRules(config, candidate, params)); + } + + @Test + public void testHasUnappliedDeletionRules_FilterNotNegated_ReturnsFalse() + { + InlineSchemaDataSourceCompactionConfig config = InlineSchemaDataSourceCompactionConfig.builder() + .forDataSource(TestDataSource.WIKI) + .withTransformSpec(new CompactionTransformSpec(new SelectorDimFilter("country", "US", null), null)) + .build(); + CompactionCandidate candidate = createCandidateWithFingerprints("fp1"); + CompactionJobParams params = createParams(); + + Assertions.assertFalse(optimizer.hasUnappliedDeletionRules(config, candidate, params)); + } + + @Test + public void testHasUnappliedDeletionRules_AllAppliedButVirtualColumnsDiffer_ReturnsFalse() + { + DimFilter ruleA = new SelectorDimFilter("vc_deletion", "x", null); + DimFilter ruleB = new SelectorDimFilter("country", "UK", null); + NotDimFilter deletionFilter = new NotDimFilter(new OrDimFilter(Arrays.asList(ruleA, ruleB))); + + // Both deletion rules applied, carrying only the deletion virtual column. + CompactionState appliedState = CompactionState.builder() + .transformSpec(new CompactionTransformSpec( + deletionFilter, + VirtualColumns.create( + new ExpressionVirtualColumn("vc_deletion", "col2 + 2", ColumnType.LONG, TestExprMacroTable.INSTANCE) + ) + )) + .indexSpec(IndexSpec.getDefault()) + .build(); + indexingStateStorage.upsertIndexingState(TestDataSource.WIKI, "fp1", appliedState, DateTimes.nowUtc()); + syncCacheFromManager(); + + // Same deletion rules, but the desired config adds a partitioning virtual column on top. + VirtualColumns desiredVcs = VirtualColumns.create( + new ExpressionVirtualColumn("vc_deletion", "col2 + 2", ColumnType.LONG, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("vc_partition", "col1 + 1", ColumnType.LONG, TestExprMacroTable.INSTANCE) + ); + CompactionCandidate candidate = createCandidateWithFingerprints("fp1"); + InlineSchemaDataSourceCompactionConfig config = createConfigWithFilter(deletionFilter, desiredVcs); + CompactionJobParams params = createParams(); + + Assertions.assertFalse(optimizer.hasUnappliedDeletionRules(config, candidate, params)); + } + /** * Helper to sync the cache with states stored in the manager (for tests that persist states). */ diff --git a/server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java b/server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java index f08997a2c6a2..a53fc5f96748 100644 --- a/server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java +++ b/server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java @@ -56,4 +56,31 @@ Eligibility checkEligibilityForCompaction( CompactionTaskStatus latestTaskStatus ); + /** + * Like {@link #checkEligibilityForCompaction} but for an interval that must be compacted regardless + * of any minimum interval-size criteria the policy enforces (e.g. an interval with unapplied deletion + * rules). Implementations skip the size gates but still choose the full-vs-minor compaction mode. The + * default upgrades a rejection to a {@link Eligibility#FULL full} compaction. + */ + default Eligibility checkEligibilityForMandatoryCompaction( + CompactionCandidate candidate, + CompactionTaskStatus latestTaskStatus + ) + { + final Eligibility eligibility = checkEligibilityForCompaction(candidate, latestTaskStatus); + return eligibility.isEligible() ? eligibility : Eligibility.FULL; + } + + /** + * Whether intervals with unapplied cascading reindexing deletion rules should be compacted even when + * they fail this policy's normal eligibility criteria (see {@link #checkEligibilityForMandatoryCompaction}). + * This applies only to cascading reindexing supervisors and their deletion rules; it has no + * effect on any other compaction. Defaults to false, so deletion rules otherwise ride along with + * regular threshold-gated compaction; operators who must apply deletions for compliance can opt in. + */ + default boolean isForcePendingDeletionCompaction() + { + return false; + } + } diff --git a/server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java b/server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java index b8b754e718cd..89196a86d773 100644 --- a/server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java +++ b/server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java @@ -51,6 +51,34 @@ public class MostFragmentedIntervalFirstPolicy extends BaseCandidateSearchPolicy private final HumanReadableBytes maxAverageUncompactedBytesPerSegment; private final int minUncompactedBytesPercentForFullCompaction; private final int minUncompactedRowsPercentForFullCompaction; + /** + * When true, intervals with unapplied cascading reindexing deletion rules bypass this policy's + * minimum interval-size gates. See {@link CompactionCandidateSearchPolicy#isForcePendingDeletionCompaction()}. + */ + private final boolean forcePendingDeletionCompaction; + + /** + * Convenience constructor that defaults {@link #forcePendingDeletionCompaction} to false. + */ + public MostFragmentedIntervalFirstPolicy( + @Nullable Integer minUncompactedCount, + @Nullable HumanReadableBytes minUncompactedBytes, + @Nullable HumanReadableBytes maxAverageUncompactedBytesPerSegment, + @Nullable Integer minUncompactedBytesPercentForFullCompaction, + @Nullable Integer minUncompactedRowsPercentForFullCompaction, + @Nullable String priorityDatasource + ) + { + this( + minUncompactedCount, + minUncompactedBytes, + maxAverageUncompactedBytesPerSegment, + minUncompactedBytesPercentForFullCompaction, + minUncompactedRowsPercentForFullCompaction, + priorityDatasource, + null + ); + } @JsonCreator public MostFragmentedIntervalFirstPolicy( @@ -62,7 +90,8 @@ public MostFragmentedIntervalFirstPolicy( Integer minUncompactedBytesPercentForFullCompaction, @JsonProperty("minUncompactedRowsPercentForFullCompaction") @Nullable Integer minUncompactedRowsPercentForFullCompaction, - @JsonProperty("priorityDatasource") @Nullable String priorityDatasource + @JsonProperty("priorityDatasource") @Nullable String priorityDatasource, + @JsonProperty("forcePendingDeletionCompaction") @Nullable Boolean forcePendingDeletionCompaction ) { super(priorityDatasource); @@ -100,6 +129,7 @@ public MostFragmentedIntervalFirstPolicy( Configs.valueOrDefault(minUncompactedBytesPercentForFullCompaction, 0); this.minUncompactedRowsPercentForFullCompaction = Configs.valueOrDefault(minUncompactedRowsPercentForFullCompaction, 0); + this.forcePendingDeletionCompaction = Configs.valueOrDefault(forcePendingDeletionCompaction, false); } /** @@ -154,6 +184,13 @@ public int minUncompactedRowsPercentForFullCompaction() return minUncompactedRowsPercentForFullCompaction; } + @JsonProperty + @Override + public boolean isForcePendingDeletionCompaction() + { + return forcePendingDeletionCompaction; + } + @Override protected Comparator getSegmentComparator() { @@ -174,7 +211,8 @@ public boolean equals(Object o) && Objects.equals(minUncompactedBytes, policy.minUncompactedBytes) && Objects.equals(maxAverageUncompactedBytesPerSegment, policy.maxAverageUncompactedBytesPerSegment) && minUncompactedBytesPercentForFullCompaction == policy.minUncompactedBytesPercentForFullCompaction - && minUncompactedRowsPercentForFullCompaction == policy.minUncompactedRowsPercentForFullCompaction; + && minUncompactedRowsPercentForFullCompaction == policy.minUncompactedRowsPercentForFullCompaction + && forcePendingDeletionCompaction == policy.forcePendingDeletionCompaction; } @Override @@ -186,7 +224,8 @@ public int hashCode() minUncompactedBytes, maxAverageUncompactedBytesPerSegment, minUncompactedBytesPercentForFullCompaction, - minUncompactedRowsPercentForFullCompaction + minUncompactedRowsPercentForFullCompaction, + forcePendingDeletionCompaction ); } @@ -201,6 +240,7 @@ public String toString() ", minUncompactedBytesPercentForFullCompaction=" + minUncompactedBytesPercentForFullCompaction + ", minUncompactedRowsPercentForFullCompaction=" + minUncompactedRowsPercentForFullCompaction + ", priorityDataSource='" + getPriorityDatasource() + '\'' + + ", forcePendingDeletionCompaction=" + forcePendingDeletionCompaction + '}'; } @@ -242,6 +282,33 @@ public Eligibility checkEligibilityForCompaction( ); } + return determineCompactionMode(candidate); + } + + /** + * Skips the minimum interval-size gates enforced by {@link #checkEligibilityForCompaction} and keeps + * only the full-vs-minor ratio decision, since the interval must be compacted regardless of its size. + */ + @Override + public Eligibility checkEligibilityForMandatoryCompaction( + CompactionCandidate candidate, + CompactionTaskStatus latestTaskStatus + ) + { + return determineCompactionMode(candidate); + } + + /** + * Chooses full vs. minor compaction from the ratio of uncompacted data (segments needing compaction + * for any reason) to total data: minor when the bytes or rows ratio is below the configured threshold. + */ + private Eligibility determineCompactionMode(CompactionCandidate candidate) + { + final CompactionStatistics uncompacted = candidate.getUncompactedStats(); + if (uncompacted == null || uncompacted.getNumSegments() < 1) { + return Eligibility.FULL; + } + final double uncompactedBytesRatio = (double) uncompacted.getTotalBytes() / (uncompacted.getTotalBytes() + candidate.getCompactedStats().getTotalBytes()) * 100; diff --git a/server/src/test/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicyTest.java b/server/src/test/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicyTest.java index 3f958756ed61..c2a7f4e35e32 100644 --- a/server/src/test/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicyTest.java +++ b/server/src/test/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicyTest.java @@ -383,6 +383,107 @@ public void test_compactionMode_returnsFullCompaction_whenThresholdIsDefault() Assertions.assertTrue(eligibility.isEligible()); } + @Test + public void test_checkEligibilityForMandatoryCompaction_bypassesMinimumSizeGates_andReturnsFull() + { + final MostFragmentedIntervalFirstPolicy policy = new MostFragmentedIntervalFirstPolicy( + 10_000, + HumanReadableBytes.valueOf(10_000), + HumanReadableBytes.valueOf(100), + null, + null, + null + ); + + final CompactionCandidate candidate = createCandidate(1, 100L); + Assertions.assertFalse(policy.checkEligibilityForCompaction(candidate, null).isEligible()); + + final Eligibility mandatory = policy.checkEligibilityForMandatoryCompaction(candidate, null); + Assertions.assertTrue(mandatory.isEligible()); + Assertions.assertEquals(CompactionMode.ALL_SEGMENTS, mandatory.getMode()); + } + + @Test + public void test_checkEligibilityForMandatoryCompaction_appliesRatioDecision_andReturnsMinor() + { + final MostFragmentedIntervalFirstPolicy policy = new MostFragmentedIntervalFirstPolicy( + 10_000, + HumanReadableBytes.valueOf(1), + HumanReadableBytes.valueOf(10_000), + 50, + 50, + null + ); + + final CompactionStatistics compacted = CompactionStatistics.create(1200L, 400L, 10, 1L); + final CompactionStatistics uncompacted = CompactionStatistics.create(400L, 600L, 100, 1L); + final CompactionStatus status = CompactionStatus.pending(compacted, uncompacted, List.of(SEGMENT), ""); + final CompactionCandidate candidate = CompactionCandidate.from(List.of(SEGMENT), null, status); + + Assertions.assertFalse(policy.checkEligibilityForCompaction(candidate, null).isEligible()); + + // uncompacted bytes ratio 400/1600 = 25%, below the 50% threshold, so minor is selected. + final Eligibility mandatory = policy.checkEligibilityForMandatoryCompaction(candidate, null); + Assertions.assertTrue(mandatory.isEligible()); + Assertions.assertEquals(CompactionMode.UNCOMPACTED_SEGMENTS_ONLY, mandatory.getMode()); + } + + @Test + public void test_checkEligibilityForMandatoryCompaction_defaultImplementation_upgradesRejectionToFull() + { + final CompactionCandidate candidate = createCandidate(1, 100L); + + final CompactionCandidateSearchPolicy rejectingPolicy = new CompactionCandidateSearchPolicy() + { + @Override + public int compareCandidates(CompactionCandidate a, CompactionCandidate b) + { + return 0; + } + + @Override + public Eligibility checkEligibilityForCompaction(CompactionCandidate c, CompactionTaskStatus s) + { + return Eligibility.fail("not eligible"); + } + }; + + Assertions.assertEquals( + Eligibility.FULL, + rejectingPolicy.checkEligibilityForMandatoryCompaction(candidate, null) + ); + } + + @Test + public void test_isForcePendingDeletionCompaction_defaultsToFalse() + { + Assertions.assertFalse( + new MostFragmentedIntervalFirstPolicy(null, null, null, null, null, null).isForcePendingDeletionCompaction() + ); + Assertions.assertFalse(new NewestSegmentFirstPolicy(null).isForcePendingDeletionCompaction()); + } + + @Test + public void test_serde_forcePendingDeletionCompaction() throws IOException + { + final MostFragmentedIntervalFirstPolicy policy = new MostFragmentedIntervalFirstPolicy( + 1, + HumanReadableBytes.valueOf(2), + HumanReadableBytes.valueOf(3), + 50, + 51, + "foo", + true + ); + Assertions.assertTrue(policy.isForcePendingDeletionCompaction()); + + final DefaultObjectMapper mapper = new DefaultObjectMapper(); + final CompactionCandidateSearchPolicy roundTripped = + mapper.readValue(mapper.writeValueAsString(policy), CompactionCandidateSearchPolicy.class); + Assertions.assertEquals(policy, roundTripped); + Assertions.assertTrue(roundTripped.isForcePendingDeletionCompaction()); + } + private CompactionCandidate createCandidate(int numSegments, long avgSizeBytes) { final CompactionStatistics dummyCompactedStats = CompactionStatistics.create(1L, null, 1L, 1L); From 6cf7107edfe4b92ca03de85dc8869562fb4c0f95 Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Tue, 30 Jun 2026 11:57:21 -0500 Subject: [PATCH 2/4] self review self review follow up Few more cleanups for self review --- docs/data-management/cascading-reindexing.md | 4 +- .../compact/CompactionSupervisorTest.java | 4 +- .../indexing/KafkaClusterMetricsTest.java | 2 +- .../CompactionConfigBasedJobTemplate.java | 2 +- .../compact/ReindexingConfigOptimizer.java | 4 +- .../CompactionCandidateSearchPolicy.java | 18 ++++----- .../MostFragmentedIntervalFirstPolicy.java | 36 ++++++------------ ...MostFragmentedIntervalFirstPolicyTest.java | 38 +++++++++++++++---- 8 files changed, 59 insertions(+), 49 deletions(-) diff --git a/docs/data-management/cascading-reindexing.md b/docs/data-management/cascading-reindexing.md index 550ecebc3190..b88faf1de0a8 100644 --- a/docs/data-management/cascading-reindexing.md +++ b/docs/data-management/cascading-reindexing.md @@ -274,9 +274,9 @@ Example: **Applying deletions regardless of compaction thresholds:** -When the cluster compaction policy is [`mostFragmentedFirst`](../api-reference/automatic-compaction-api.md#compaction-policy-mostfragmentedfirst), an interval must meet minimum size thresholds (such as `minUncompactedCount`) before it is compacted. An interval that recently crossed a deletion rule's `olderThan` threshold may have too few uncompacted segments to qualify, which delays applying the deletion. +When the cluster compaction policy is [`mostFragmentedFirst`](../api-reference/automatic-compaction-api.md#compaction-policy-mostfragmentedfirst), an interval must meet various minimum thresholds (such as `minUncompactedCount`) before it is compacted. This intentionally avoids compacting intervals with low fragmentation. -Operators who apply deletion rules for data compliance can set `forcePendingDeletionCompaction: true` on the policy. Any interval that has deletion rules not yet applied to all of its segments is then compacted regardless of those thresholds, while still using the policy's full-vs-minor decision. This applies only to cascading reindexing deletion rules; it does not change eligibility for any other compaction. The setting defaults to `false`, so by default deletions wait until the interval qualifies for compaction on its own. +Operators who apply deletion rules for data compliance (or are generally interested in eagerly applying their deletion rules) can set `forcePendingDeletionCompaction: true` on the policy. Any interval that has deletion rules not yet applied to all of its segments is then compacted regardless of those thresholds, while still using the policy's full-vs-minor decision. This applies only to cascading reindexing deletion rules; it does not change eligibility for any other compaction. The setting defaults to `false`, so by default deletions wait until the interval qualifies for compaction on its own. #### Index spec rules diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java index b6cca782795d..3b44f0d2d16d 100644 --- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java @@ -1288,12 +1288,12 @@ public static List getPolicyAndPartition() return List.of( new Object[]{ // decides minor compaction based on bytes percent - new MostFragmentedIntervalFirstPolicy(2, new HumanReadableBytes("1KiB"), null, 80, null, null), + new MostFragmentedIntervalFirstPolicy(2, new HumanReadableBytes("1KiB"), null, 80, null, null, null), new DimensionRangePartitionsSpec(null, 10_000, List.of("page"), false) }, new Object[]{ // decides minor compaction based on rows percent - new MostFragmentedIntervalFirstPolicy(2, new HumanReadableBytes("1KiB"), null, null, 51, null), + new MostFragmentedIntervalFirstPolicy(2, new HumanReadableBytes("1KiB"), null, null, 51, null, null), new DynamicPartitionsSpec(null, null) } ); diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaClusterMetricsTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaClusterMetricsTest.java index bc7385e6bf6e..dfa06febe621 100644 --- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaClusterMetricsTest.java +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaClusterMetricsTest.java @@ -147,7 +147,7 @@ public static Stream getCompactionSupervisorTestParams() ), Arguments.of( CompactionEngine.MSQ, - new MostFragmentedIntervalFirstPolicy(1, HumanReadableBytes.valueOf(1), null, null, 80, null) + new MostFragmentedIntervalFirstPolicy(1, HumanReadableBytes.valueOf(1), null, null, 80, null, null) ) ); } diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java index 0a589b7a90e4..3542bc9dee3d 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java @@ -100,7 +100,7 @@ public List createCompactionJobs( final CompactionTaskStatus latestTaskStatus = params.getLatestTaskStatus(candidate); Eligibility eligibility = policy.checkEligibilityForCompaction(candidate, latestTaskStatus); - if (policy.isForcePendingDeletionCompaction() + if (policy.isForceMandatoryCompactionEnabled() && !eligibility.isEligible() && configOptimizer.hasUnappliedDeletionRules(config, candidate, params)) { // When the operator has opted in, a cascading reindexing interval with unapplied deletion rules diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingConfigOptimizer.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingConfigOptimizer.java index d76a46c2dfb0..93c3f4b23ce6 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingConfigOptimizer.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingConfigOptimizer.java @@ -49,8 +49,8 @@ DataSourceCompactionConfig optimizeConfig( /** * Whether {@code candidate} has deletion rules not yet applied to all of its segments. The job - * template uses this to force compaction eligibility for intervals that must be reindexed for data - * compliance, regardless of the policy's minimum interval-size thresholds. Defaults to false. + * template uses this to force compaction eligibility for intervals that must be reindexed to apply + * deletion rules, regardless of the policy's minimum interval-size thresholds. Defaults to false. */ default boolean hasUnappliedDeletionRules( DataSourceCompactionConfig config, diff --git a/server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java b/server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java index a53fc5f96748..8bb7b2844e5f 100644 --- a/server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java +++ b/server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java @@ -58,9 +58,9 @@ Eligibility checkEligibilityForCompaction( /** * Like {@link #checkEligibilityForCompaction} but for an interval that must be compacted regardless - * of any minimum interval-size criteria the policy enforces (e.g. an interval with unapplied deletion - * rules). Implementations skip the size gates but still choose the full-vs-minor compaction mode. The - * default upgrades a rejection to a {@link Eligibility#FULL full} compaction. + * of any implementation-specific eligibility criteria. By default, this method calls + * {@link #checkEligibilityForCompaction} and by default upgrades a rejection to a {@link Eligibility#FULL full} + * compaction. */ default Eligibility checkEligibilityForMandatoryCompaction( CompactionCandidate candidate, @@ -72,13 +72,13 @@ default Eligibility checkEligibilityForMandatoryCompaction( } /** - * Whether intervals with unapplied cascading reindexing deletion rules should be compacted even when - * they fail this policy's normal eligibility criteria (see {@link #checkEligibilityForMandatoryCompaction}). - * This applies only to cascading reindexing supervisors and their deletion rules; it has no - * effect on any other compaction. Defaults to false, so deletion rules otherwise ride along with - * regular threshold-gated compaction; operators who must apply deletions for compliance can opt in. + * Whether an interval that is below this policy's normal eligibility criteria should still be compacted + * via {@link #checkEligibilityForMandatoryCompaction} when an external trigger requires it. The caller + * decides what constitutes such a trigger; today the only one is a cascading reindexing interval with + * deletion rules not yet applied to all of its segments. Defaults to false, so mandatory compaction is + * never forced unless a policy opts in. */ - default boolean isForcePendingDeletionCompaction() + default boolean isForceMandatoryCompactionEnabled() { return false; } diff --git a/server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java b/server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java index 89196a86d773..2bc64e588d36 100644 --- a/server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java +++ b/server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java @@ -53,33 +53,10 @@ public class MostFragmentedIntervalFirstPolicy extends BaseCandidateSearchPolicy private final int minUncompactedRowsPercentForFullCompaction; /** * When true, intervals with unapplied cascading reindexing deletion rules bypass this policy's - * minimum interval-size gates. See {@link CompactionCandidateSearchPolicy#isForcePendingDeletionCompaction()}. + * minimum interval-size gates. See {@link CompactionCandidateSearchPolicy#isForceMandatoryCompactionEnabled()}. */ private final boolean forcePendingDeletionCompaction; - /** - * Convenience constructor that defaults {@link #forcePendingDeletionCompaction} to false. - */ - public MostFragmentedIntervalFirstPolicy( - @Nullable Integer minUncompactedCount, - @Nullable HumanReadableBytes minUncompactedBytes, - @Nullable HumanReadableBytes maxAverageUncompactedBytesPerSegment, - @Nullable Integer minUncompactedBytesPercentForFullCompaction, - @Nullable Integer minUncompactedRowsPercentForFullCompaction, - @Nullable String priorityDatasource - ) - { - this( - minUncompactedCount, - minUncompactedBytes, - maxAverageUncompactedBytesPerSegment, - minUncompactedBytesPercentForFullCompaction, - minUncompactedRowsPercentForFullCompaction, - priorityDatasource, - null - ); - } - @JsonCreator public MostFragmentedIntervalFirstPolicy( @JsonProperty("minUncompactedCount") @Nullable Integer minUncompactedCount, @@ -185,12 +162,21 @@ public int minUncompactedRowsPercentForFullCompaction() } @JsonProperty - @Override public boolean isForcePendingDeletionCompaction() { return forcePendingDeletionCompaction; } + /** + * Forced when the operator has opted into compacting intervals with pending deletion rules. Other + * triggers can be OR'd in here in the future without affecting how this policy is serialized. + */ + @Override + public boolean isForceMandatoryCompactionEnabled() + { + return forcePendingDeletionCompaction; + } + @Override protected Comparator getSegmentComparator() { diff --git a/server/src/test/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicyTest.java b/server/src/test/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicyTest.java index c2a7f4e35e32..266c70ca5c37 100644 --- a/server/src/test/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicyTest.java +++ b/server/src/test/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicyTest.java @@ -40,7 +40,7 @@ public class MostFragmentedIntervalFirstPolicyTest public void test_thresholdValues_ofDefaultPolicy() { final MostFragmentedIntervalFirstPolicy policy = - new MostFragmentedIntervalFirstPolicy(null, null, null, null, null, null); + new MostFragmentedIntervalFirstPolicy(null, null, null, null, null, null, null); Assertions.assertEquals(100, policy.getMinUncompactedCount()); Assertions.assertEquals(new HumanReadableBytes("10MiB"), policy.getMinUncompactedBytes()); Assertions.assertEquals(new HumanReadableBytes("2GiB"), policy.getMaxAverageUncompactedBytesPerSegment()); @@ -57,6 +57,7 @@ public void test_checkEligibilityForCompaction_fails_ifUncompactedCountLessThanC HumanReadableBytes.valueOf(10_000), null, null, + null, null ); @@ -82,6 +83,7 @@ public void test_checkEligibilityForCompaction_fails_ifUncompactedBytesLessThanC HumanReadableBytes.valueOf(10_000), null, null, + null, null ); @@ -107,6 +109,7 @@ public void test_checkEligibilityForCompaction_fails_ifAvgSegmentSizeGreaterThan maxAvgSegmentSize, null, null, + null, null ); @@ -131,6 +134,7 @@ public void test_policy_favorsIntervalWithMoreUncompactedSegments_ifTotalBytesIs HumanReadableBytes.valueOf(10_000), null, null, + null, null ); @@ -153,6 +157,7 @@ public void test_policy_favorsIntervalWithMoreUncompactedSegments_ifAverageSizeI HumanReadableBytes.valueOf(10_000), null, null, + null, null ); @@ -175,6 +180,7 @@ public void test_policy_favorsIntervalWithSmallerSegments_ifCountIsEqual() HumanReadableBytes.valueOf(10_000), null, null, + null, null ); @@ -197,6 +203,7 @@ public void test_compareCandidates_returnsZeroIfSegmentCountAndAvgSizeScaleEquiv HumanReadableBytes.valueOf(100), null, null, + null, null ); @@ -228,7 +235,8 @@ public void test_serde_allFieldsSet() throws IOException HumanReadableBytes.valueOf(3), 50, 51, - "foo" + "foo", + null ); final DefaultObjectMapper mapper = new DefaultObjectMapper(); final CompactionCandidateSearchPolicy policy2 = @@ -240,7 +248,7 @@ public void test_serde_allFieldsSet() throws IOException public void test_serde_noFieldsSet() throws IOException { final MostFragmentedIntervalFirstPolicy policy = - new MostFragmentedIntervalFirstPolicy(null, null, null, null, null, null); + new MostFragmentedIntervalFirstPolicy(null, null, null, null, null, null, null); final DefaultObjectMapper mapper = new DefaultObjectMapper(); final CompactionCandidateSearchPolicy policy2 = mapper.readValue(mapper.writeValueAsString(policy), CompactionCandidateSearchPolicy.class); @@ -257,6 +265,7 @@ public void test_compactionMode_returnsMinorCompactionMode_whenBytePercentageBel HumanReadableBytes.valueOf(10_000), 50, 50, + null, null ); @@ -282,6 +291,7 @@ public void test_compactionMode_returnsFullCompaction_whenBytePercentageAboveThr HumanReadableBytes.valueOf(10_000), 50, 50, + null, null ); @@ -309,6 +319,7 @@ public void test_compactionMode_returnsMinorCompaction_whenRowPercentageBelowThr HumanReadableBytes.valueOf(10_000), 50, 50, + null, null ); @@ -336,6 +347,7 @@ public void test_compactionMode_returnsFullCompaction_whenByteAndRowPercentageBo HumanReadableBytes.valueOf(10_000), 50, 50, + null, null ); @@ -364,6 +376,7 @@ public void test_compactionMode_returnsFullCompaction_whenThresholdIsDefault() HumanReadableBytes.valueOf(10_000), null, null, + null, null ); @@ -392,6 +405,7 @@ public void test_checkEligibilityForMandatoryCompaction_bypassesMinimumSizeGates HumanReadableBytes.valueOf(100), null, null, + null, null ); @@ -412,6 +426,7 @@ public void test_checkEligibilityForMandatoryCompaction_appliesRatioDecision_and HumanReadableBytes.valueOf(10_000), 50, 50, + null, null ); @@ -455,12 +470,20 @@ public Eligibility checkEligibilityForCompaction(CompactionCandidate c, Compacti } @Test - public void test_isForcePendingDeletionCompaction_defaultsToFalse() + public void test_isForceMandatoryCompactionEnabled_defaultsToFalse() { Assertions.assertFalse( - new MostFragmentedIntervalFirstPolicy(null, null, null, null, null, null).isForcePendingDeletionCompaction() + new MostFragmentedIntervalFirstPolicy( + null, + null, + null, + null, + null, + null, + null + ).isForceMandatoryCompactionEnabled() ); - Assertions.assertFalse(new NewestSegmentFirstPolicy(null).isForcePendingDeletionCompaction()); + Assertions.assertFalse(new NewestSegmentFirstPolicy(null).isForceMandatoryCompactionEnabled()); } @Test @@ -476,12 +499,13 @@ public void test_serde_forcePendingDeletionCompaction() throws IOException true ); Assertions.assertTrue(policy.isForcePendingDeletionCompaction()); + Assertions.assertTrue(policy.isForceMandatoryCompactionEnabled()); final DefaultObjectMapper mapper = new DefaultObjectMapper(); final CompactionCandidateSearchPolicy roundTripped = mapper.readValue(mapper.writeValueAsString(policy), CompactionCandidateSearchPolicy.class); Assertions.assertEquals(policy, roundTripped); - Assertions.assertTrue(roundTripped.isForcePendingDeletionCompaction()); + Assertions.assertTrue(roundTripped.isForceMandatoryCompactionEnabled()); } private CompactionCandidate createCandidate(int numSegments, long avgSizeBytes) From 2d6620dad159c7838482a35165609e34e6fe86a6 Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Tue, 30 Jun 2026 16:02:11 -0500 Subject: [PATCH 3/4] Add a log when we force a compaction --- .../compact/CompactionConfigBasedJobTemplate.java | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java index 3542bc9dee3d..ba77f62ea2d1 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionConfigBasedJobTemplate.java @@ -27,6 +27,7 @@ import org.apache.druid.indexing.input.DruidInputSource; import org.apache.druid.java.util.common.Intervals; import org.apache.druid.java.util.common.granularity.Granularity; +import org.apache.druid.java.util.common.logger.Logger; import org.apache.druid.server.compaction.CompactionCandidate; import org.apache.druid.server.compaction.CompactionCandidateSearchPolicy; import org.apache.druid.server.compaction.CompactionSlotManager; @@ -52,6 +53,8 @@ */ public class CompactionConfigBasedJobTemplate implements CompactionJobTemplate { + private static final Logger log = new Logger(CompactionConfigBasedJobTemplate.class); + private final DataSourceCompactionConfig config; private final ReindexingConfigOptimizer configOptimizer; @@ -106,7 +109,16 @@ public List createCompactionJobs( // When the operator has opted in, a cascading reindexing interval with unapplied deletion rules // must be reindexed for compliance even when it is below the policy's size thresholds; bypass // those gates but keep its full-vs-minor decision. + final String policyRejectionReason = eligibility.getReason(); eligibility = policy.checkEligibilityForMandatoryCompaction(candidate, latestTaskStatus); + log.info( + "Forcing compaction of interval[%s] for dataSource[%s] in mode[%s] because it has unapplied" + + " deletion rules, even though it is not normally eligible[%s].", + candidate.getUmbrellaInterval(), + config.getDataSource(), + eligibility.getMode(), + policyRejectionReason + ); } if (!eligibility.isEligible()) { continue; From 6eb4cbc5439e0dbee3372f56190ae8865bcb16f0 Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Wed, 1 Jul 2026 09:14:34 -0500 Subject: [PATCH 4/4] Fix bug in initial impl that would mis compactions --- .../ReindexingDeletionRuleOptimizer.java | 17 ++++++----- .../ReindexingDeletionRuleOptimizerTest.java | 28 +++++++++++++++++++ 2 files changed, 38 insertions(+), 7 deletions(-) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizer.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizer.java index 7b96cf697a11..9070e2fb7dc7 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizer.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizer.java @@ -39,7 +39,6 @@ import java.util.Collections; import java.util.HashSet; import java.util.List; -import java.util.Objects; import java.util.Set; import java.util.stream.Collectors; @@ -146,16 +145,20 @@ private NotDimFilter computeRequiredSetOfFilterRulesForCandidate( expectedFilters = ((OrDimFilter) expectedFilter.getField()).getFields(); } + // If any segment has no fingerprint, we cannot prove the deletion rules were applied to it, so we + // cannot prune anything: report every rule as unapplied. This also subsumes the all-null case (no + // segment can prove application). Only when every segment has a fingerprint do we compare applied + // filters against expected. + final boolean anySegmentMissingFingerprint = candidateSegments.getSegments().stream() + .anyMatch(s -> s.getIndexingStateFingerprint() == null); + if (anySegmentMissingFingerprint) { + return expectedFilter; + } + Set uniqueFingerprints = candidateSegments.getSegments().stream() .map(DataSegment::getIndexingStateFingerprint) - .filter(Objects::nonNull) .collect(Collectors.toSet()); - if (uniqueFingerprints.isEmpty()) { - // no fingerprints means that no candidate segments have transforms to compare against. Return all filters eagerly. - return expectedFilter; - } - Set unappliedRules = new HashSet<>(); for (String fingerprint : uniqueFingerprints) { diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizerTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizerTest.java index dce3b5ab1c37..a3509c2fc184 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizerTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/compact/ReindexingDeletionRuleOptimizerTest.java @@ -679,6 +679,34 @@ public void testHasUnappliedDeletionRules_AllApplied_ReturnsFalse() Assertions.assertFalse(optimizer.hasUnappliedDeletionRules(config, candidate, params)); } + @Test + public void testHasUnappliedDeletionRules_AppliedFingerprintPlusNullFingerprint_ReturnsTrue() + { + DimFilter filterA = new SelectorDimFilter("country", "US", null); + + // fp1 already has the deletion rule applied... + CompactionState state = createStateWithSingleFilter(filterA); + indexingStateStorage.upsertIndexingState(TestDataSource.WIKI, "fp1", state, DateTimes.nowUtc()); + syncCacheFromManager(); + + // ...but the candidate also contains a never-compacted (null fingerprint) segment that has not, so the + // rule is unapplied for at least one segment and this must return true. + List segments = new ArrayList<>(); + segments.add(DataSegment.builder(WIKI_SEGMENT).indexingStateFingerprint("fp1").build()); + segments.add(DataSegment.builder(WIKI_SEGMENT).indexingStateFingerprint(null).build()); + CompactionCandidate candidate = CompactionCandidate.from( + segments, + null, + CompactionStatus.pending("segments need compaction") + ); + + NotDimFilter expectedFilter = new NotDimFilter(filterA); + InlineSchemaDataSourceCompactionConfig config = createConfigWithFilter(expectedFilter, null); + CompactionJobParams params = createParams(); + + Assertions.assertTrue(optimizer.hasUnappliedDeletionRules(config, candidate, params)); + } + @Test public void testHasUnappliedDeletionRules_MissingState_ReturnsTrue() {