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..b88faf1de0a8 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 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 (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 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..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 @@ -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() { @@ -1235,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 1a8dd95179b5..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,8 +27,11 @@ 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; +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; @@ -50,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; @@ -94,10 +99,27 @@ 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.isForceMandatoryCompactionEnabled() + && !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. + 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; } 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..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 @@ -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 to apply + * deletion rules, 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..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; @@ -95,6 +94,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}. *

@@ -121,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 100349d96409..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 @@ -629,6 +629,176 @@ 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_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() + { + 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..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 @@ -56,4 +56,31 @@ Eligibility checkEligibilityForCompaction( CompactionTaskStatus latestTaskStatus ); + /** + * Like {@link #checkEligibilityForCompaction} but for an interval that must be compacted regardless + * 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, + CompactionTaskStatus latestTaskStatus + ) + { + final Eligibility eligibility = checkEligibilityForCompaction(candidate, latestTaskStatus); + return eligibility.isEligible() ? eligibility : Eligibility.FULL; + } + + /** + * 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 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 b8b754e718cd..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 @@ -51,6 +51,11 @@ 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#isForceMandatoryCompactionEnabled()}. + */ + private final boolean forcePendingDeletionCompaction; @JsonCreator public MostFragmentedIntervalFirstPolicy( @@ -62,7 +67,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 +106,7 @@ public MostFragmentedIntervalFirstPolicy( Configs.valueOrDefault(minUncompactedBytesPercentForFullCompaction, 0); this.minUncompactedRowsPercentForFullCompaction = Configs.valueOrDefault(minUncompactedRowsPercentForFullCompaction, 0); + this.forcePendingDeletionCompaction = Configs.valueOrDefault(forcePendingDeletionCompaction, false); } /** @@ -154,6 +161,22 @@ public int minUncompactedRowsPercentForFullCompaction() return minUncompactedRowsPercentForFullCompaction; } + @JsonProperty + 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() { @@ -174,7 +197,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 +210,8 @@ public int hashCode() minUncompactedBytes, maxAverageUncompactedBytesPerSegment, minUncompactedBytesPercentForFullCompaction, - minUncompactedRowsPercentForFullCompaction + minUncompactedRowsPercentForFullCompaction, + forcePendingDeletionCompaction ); } @@ -201,6 +226,7 @@ public String toString() ", minUncompactedBytesPercentForFullCompaction=" + minUncompactedBytesPercentForFullCompaction + ", minUncompactedRowsPercentForFullCompaction=" + minUncompactedRowsPercentForFullCompaction + ", priorityDataSource='" + getPriorityDatasource() + '\'' + + ", forcePendingDeletionCompaction=" + forcePendingDeletionCompaction + '}'; } @@ -242,6 +268,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..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 ); @@ -383,6 +396,118 @@ 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, + 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, + 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_isForceMandatoryCompactionEnabled_defaultsToFalse() + { + Assertions.assertFalse( + new MostFragmentedIntervalFirstPolicy( + null, + null, + null, + null, + null, + null, + null + ).isForceMandatoryCompactionEnabled() + ); + Assertions.assertFalse(new NewestSegmentFirstPolicy(null).isForceMandatoryCompactionEnabled()); + } + + @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()); + 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.isForceMandatoryCompactionEnabled()); + } + private CompactionCandidate createCandidate(int numSegments, long avgSizeBytes) { final CompactionStatistics dummyCompactedStats = CompactionStatistics.create(1L, null, 1L, 1L);