diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CascadingReindexingTemplate.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CascadingReindexingTemplate.java index 2f22f20af136..9471b4c0ee9d 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CascadingReindexingTemplate.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CascadingReindexingTemplate.java @@ -101,6 +101,7 @@ public class CascadingReindexingTemplate implements CompactionJobTemplate, DataS private final int taskPriority; private final long inputSegmentSizeBytes; private final Period skipOffsetFromLatest; + private final Period skipOffsetFromEarliest; private final Period skipOffsetFromNow; private final Granularity defaultSegmentGranularity; private final PartitionsSpec defaultPartitionsSpec; @@ -118,6 +119,7 @@ public CascadingReindexingTemplate( @JsonProperty("ruleProvider") ReindexingRuleProvider ruleProvider, @JsonProperty("taskContext") @Nullable Map taskContext, @JsonProperty("skipOffsetFromLatest") @Nullable Period skipOffsetFromLatest, + @JsonProperty("skipOffsetFromEarliest") @Nullable Period skipOffsetFromEarliest, @JsonProperty("skipOffsetFromNow") @Nullable Period skipOffsetFromNow, @JsonProperty("defaultSegmentGranularity") Granularity defaultSegmentGranularity, @JsonProperty("defaultPartitionsSpec") PartitionsSpec defaultPartitionsSpec, @@ -152,11 +154,15 @@ public CascadingReindexingTemplate( } this.tuningConfig = tuningConfig; - if (skipOffsetFromNow != null && skipOffsetFromLatest != null) { - throw InvalidInput.exception("Cannot set both skipOffsetFromNow and skipOffsetFromLatest"); + int skipOffsetCount = (skipOffsetFromNow != null ? 1 : 0) + + (skipOffsetFromLatest != null ? 1 : 0) + + (skipOffsetFromEarliest != null ? 1 : 0); + if (skipOffsetCount > 1) { + throw InvalidInput.exception("Cannot set more than one of: skipOffsetFromNow, skipOffsetFromLatest, skipOffsetFromEarliest"); } this.skipOffsetFromNow = skipOffsetFromNow; this.skipOffsetFromLatest = skipOffsetFromLatest; + this.skipOffsetFromEarliest = skipOffsetFromEarliest; this.defaultPartitioningRule = ReindexingPartitioningRule.syntheticRule( defaultSegmentGranularity, @@ -220,6 +226,14 @@ public Period getSkipOffsetFromLatest() return skipOffsetFromLatest; } + @Override + @JsonProperty + @Nullable + public Period getSkipOffsetFromEarliest() + { + return skipOffsetFromEarliest; + } + @JsonProperty @Nullable private Period getSkipOffsetFromNow() @@ -366,7 +380,7 @@ public List createCompactionJobs( } // Skip offsets, if configured, can result in needing to truncate a search interval. If the truncation makes the interval invalid, skip it. - if ((skipOffsetFromNow != null || skipOffsetFromLatest != null) && + if ((skipOffsetFromNow != null || skipOffsetFromLatest != null || skipOffsetFromEarliest != null) && intervalEndsAfter(reindexingInterval, adjustedTimelineInterval.getEnd())) { DateTime alignedEnd = intervalInfo.getGranularity().bucketStart(adjustedTimelineInterval.getEnd()); @@ -422,13 +436,14 @@ protected CompactionJobTemplate createJobTemplateForInterval( } /** - * Applies the configured skip offset to an interval by adjusting its end time. Uses either - * skipOffsetFromNow (relative to reference time) or skipOffsetFromLatest (relative to interval end). - * Returns null if the adjusted end would be before the interval start. + * Applies the configured skip offset to an interval by adjusting its start or end time. Uses either + * skipOffsetFromNow (relative to reference time), skipOffsetFromLatest (relative to interval end), + * or skipOffsetFromEarliest (relative to interval start). + * Returns null if the adjusted interval would be invalid. * * @param interval the interval to adjust * @param skipFromNowReferenceTime the reference time for skipOffsetFromNow calculation - * @return the interval with adjusted end time, or null if the result would be invalid + * @return the interval with adjusted boundaries, or null if the result would be invalid */ @Nullable private Interval applySkipOffset( @@ -436,16 +451,21 @@ private Interval applySkipOffset( DateTime skipFromNowReferenceTime ) { + DateTime maybeAdjustedStart = interval.getStart(); DateTime maybeAdjustedEnd = interval.getEnd(); + if (skipOffsetFromNow != null) { maybeAdjustedEnd = skipFromNowReferenceTime.minus(skipOffsetFromNow); } else if (skipOffsetFromLatest != null) { maybeAdjustedEnd = maybeAdjustedEnd.minus(skipOffsetFromLatest); + } else if (skipOffsetFromEarliest != null) { + maybeAdjustedStart = maybeAdjustedStart.plus(skipOffsetFromEarliest); } - if (maybeAdjustedEnd.isBefore(interval.getStart())) { + + if (maybeAdjustedEnd.isBefore(maybeAdjustedStart) || maybeAdjustedEnd.equals(maybeAdjustedStart)) { return null; } else { - return new Interval(interval.getStart(), maybeAdjustedEnd); + return new Interval(maybeAdjustedStart, maybeAdjustedEnd); } } @@ -458,7 +478,8 @@ private InlineSchemaDataSourceCompactionConfig.Builder createBaseBuilder() .withInputSegmentSizeBytes(inputSegmentSizeBytes) .withEngine(CompactionEngine.MSQ) .withTaskContext(taskContext) - .withSkipOffsetFromLatest(Period.ZERO); // We handle skip offsets at the timeline level, we know we want to cover the entirety of the interval + .withSkipOffsetFromLatest(Period.ZERO) // We handle skip offsets at the timeline level + .withSkipOffsetFromEarliest(Period.ZERO); // We handle skip offsets at the timeline level } /** diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/compact/CascadingReindexingTemplateTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/compact/CascadingReindexingTemplateTest.java index af8d732311e3..0c6e8feea73d 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/compact/CascadingReindexingTemplateTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/compact/CascadingReindexingTemplateTest.java @@ -109,6 +109,7 @@ public void test_serde() throws Exception ImmutableMap.of("context_key", "context_value"), null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -148,6 +149,7 @@ public void test_serde_asDataSourceCompactionConfig() throws Exception ImmutableMap.of("key", "value"), null, null, + null, Granularities.HOUR, new DynamicPartitionsSpec(5000000, null), null, @@ -185,6 +187,7 @@ public void test_createCompactionJobs_ruleProviderNotReady() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -213,6 +216,7 @@ public void test_constructor_setBothSkipOffsetStrategiesThrowsException() mockProvider, null, Period.days(7), // skipOffsetFromLatest + null, // skipOffsetFromEarliest Period.days(3), // skipOffsetFromNow Granularities.DAY, new DynamicPartitionsSpec(5000000, null), @@ -221,7 +225,7 @@ public void test_constructor_setBothSkipOffsetStrategiesThrowsException() ) ); - Assertions.assertEquals("Cannot set both skipOffsetFromNow and skipOffsetFromLatest", exception.getMessage()); + Assertions.assertTrue(exception.getMessage().contains("Cannot set more than one of")); EasyMock.verify(mockProvider); } @@ -241,6 +245,7 @@ public void test_constructor_nullDataSourceThrowsException() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -265,6 +270,7 @@ public void test_constructor_nullRuleProviderThrowsException() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -291,6 +297,7 @@ public void test_constructor_nullDefaultSegmentGranularityThrowsException() null, null, null, + null, null, // null defaultSegmentGranularity new DynamicPartitionsSpec(5000000, null), null, @@ -322,6 +329,7 @@ public void test_constructor_tuningConfigWithPartitionsSpecThrowsException() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -557,6 +565,7 @@ public void test_generateAlignedSearchIntervals_withGranularityAlignment() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -647,6 +656,7 @@ public void test_generateAlignedSearchIntervals_withNonPartitioningRuleSplits() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -742,6 +752,7 @@ public void test_generateAlignedSearchIntervals_withNoPartitioningRules() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -838,6 +849,7 @@ public void test_generateAlignedSearchIntervals_prependIntervalForShortNonPartit null, null, null, + null, Granularities.HOUR, new DynamicPartitionsSpec(5000000, null), null, @@ -946,6 +958,7 @@ public void test_generateAlignedSearchIntervals() null, null, null, + null, Granularities.HOUR, new DynamicPartitionsSpec(5000000, null), null, @@ -1017,6 +1030,7 @@ public void test_generateAlignedSearchIntervals_noRulesThrowsException() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -1084,6 +1098,7 @@ public void test_generateAlignedSearchIntervals_splitPointSnapsToExistingBoundar null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -1152,6 +1167,7 @@ public void test_generateAlignedSearchIntervals_prependAlignmentDoesNotExtendTim null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -1224,6 +1240,7 @@ public void test_generateAlignedSearchIntervals_duplicateSplitPointsFiltered() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -1288,6 +1305,7 @@ public void test_generateAlignedSearchIntervals_singleRuleOnly() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -1355,6 +1373,7 @@ public void test_generateAlignedSearchIntervals_zeroPeriodRuleAppliesImmediately null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -1447,6 +1466,7 @@ public void test_generateAlignedSearchIntervals_zeroPeriodRuleWithOtherRules() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -1519,6 +1539,7 @@ public void test_generateAlignedSearchIntervals_failsWhenDefaultGranularityIsCoa null, null, null, + null, Granularities.MONTH, // MONTH is coarser than HOUR! new DynamicPartitionsSpec(5000000, null), null, @@ -1582,6 +1603,7 @@ public void test_generateAlignedSearchIntervals_failsWhenOlderRuleHasFinerGranul null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), null, @@ -1629,6 +1651,7 @@ public void test_generateAlignedSearchIntervals_defaultPartitioningVirtualColumn null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), defaultVCs, @@ -1689,6 +1712,7 @@ public void test_generateAlignedSearchIntervals_defaultPartitioningVirtualColumn null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), defaultVCs, @@ -1746,6 +1770,7 @@ public void test_serde_withDefaultPartitioningVirtualColumns() throws Exception ImmutableMap.of("context_key", "context_value"), null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(5000000, null), defaultVCs, @@ -1788,6 +1813,7 @@ public void test_validate_returnsValid_withDynamicPartitionsSpec() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(null, null), null, @@ -1809,6 +1835,7 @@ public void test_validate_returnsInvalid_withHashedPartitionsSpec() null, null, null, + null, Granularities.DAY, new HashedPartitionsSpec(null, 3, null), null, @@ -1834,6 +1861,7 @@ public void test_validate_returnsInvalid_withMaxTotalRows() null, null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(null, 1000L), null, @@ -1859,6 +1887,7 @@ public void test_validate_returnsInvalid_withOneMaxNumTasks() Collections.singletonMap(ClientMSQContext.CTX_MAX_NUM_TASKS, 1), null, null, + null, Granularities.DAY, new DynamicPartitionsSpec(null, null), null, diff --git a/server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java b/server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java index f818d6ca97a6..2f67e3fac9ca 100644 --- a/server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java +++ b/server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java @@ -350,26 +350,36 @@ private void findAndEnqueueSegmentsToCompact(CompactibleSegmentIterator compacti } /** - * Returns the initial searchInterval which is {@code (timeline.first().start, timeline.last().end - skipOffset)}. + * Returns the initial searchInterval which is {@code (timeline.first().start + skipOffsetFromEarliest, timeline.last().end - skipOffsetFromLatest)}. */ private List findInitialSearchInterval( SegmentTimeline timeline, @Nullable List skipIntervals ) { - final Period skipOffset = config.getSkipOffsetFromLatest(); + final Period skipOffsetFromLatest = config.getSkipOffsetFromLatest(); + final Period skipOffsetFromEarliest = config.getSkipOffsetFromEarliest(); Preconditions.checkArgument(timeline != null && !timeline.isEmpty(), "timeline should not be null or empty"); - Preconditions.checkNotNull(skipOffset, "skipOffset"); + Preconditions.checkNotNull(skipOffsetFromLatest, "skipOffsetFromLatest"); + Preconditions.checkNotNull(skipOffsetFromEarliest, "skipOffsetFromEarliest"); final TimelineObjectHolder first = Preconditions.checkNotNull(timeline.first(), "first"); final TimelineObjectHolder last = Preconditions.checkNotNull(timeline.last(), "last"); + final Interval latestSkipInterval = computeLatestSkipInterval( config.getSegmentGranularity(), last.getInterval().getEnd(), - skipOffset + skipOffsetFromLatest + ); + + final Interval earliestSkipInterval = computeEarliestSkipInterval( + config.getSegmentGranularity(), + first.getInterval().getStart(), + skipOffsetFromEarliest ); + final List allSkipIntervals - = sortAndAddSkipIntervalFromLatest(latestSkipInterval, skipIntervals); + = sortAndAddSkipIntervals(latestSkipInterval, earliestSkipInterval, skipIntervals); // Collect stats for all skipped segments for (Interval skipInterval : allSkipIntervals) { @@ -384,7 +394,9 @@ private List findInitialSearchInterval( final CompactionStatus reason; if (compactionInterval.overlaps(latestSkipInterval)) { - reason = CompactionStatus.skipped("skip offset from latest[%s]", skipOffset); + reason = CompactionStatus.skipped("skip offset from latest[%s]", skipOffsetFromLatest); + } else if (compactionInterval.overlaps(earliestSkipInterval)) { + reason = CompactionStatus.skipped("skip offset from earliest[%s]", skipOffsetFromEarliest); } else { reason = CompactionStatus.skipped("interval locked by another task"); } @@ -454,7 +466,52 @@ static Interval computeLatestSkipInterval( } } + static Interval computeEarliestSkipInterval( + @Nullable Granularity configuredSegmentGranularity, + DateTime earliestDataTimestamp, + Period skipOffsetFromEarliest + ) + { + if (configuredSegmentGranularity == null) { + return new Interval(earliestDataTimestamp, earliestDataTimestamp.plus(skipOffsetFromEarliest)); + } else { + DateTime skipFromEarliest = new DateTime(earliestDataTimestamp, earliestDataTimestamp.getZone()).plus(skipOffsetFromEarliest); + DateTime skipOffsetBucketToSegmentGranularity = configuredSegmentGranularity.bucketStart(skipFromEarliest); + return new Interval(earliestDataTimestamp, skipOffsetBucketToSegmentGranularity); + } + } + + @VisibleForTesting + static List sortAndAddSkipIntervals( + Interval skipFromLatest, + Interval skipFromEarliest, + @Nullable List skipIntervals + ) + { + final List nonNullSkipIntervals = skipIntervals == null + ? new ArrayList<>(2) + : new ArrayList<>(skipIntervals.size() + 2); + + if (skipIntervals != null) { + nonNullSkipIntervals.addAll(skipIntervals); + } + + // Add earliest skip interval if it's not empty + if (!skipFromEarliest.getStart().equals(skipFromEarliest.getEnd())) { + nonNullSkipIntervals.add(skipFromEarliest); + } + + // Add latest skip interval + nonNullSkipIntervals.add(skipFromLatest); + + // Sort all intervals + nonNullSkipIntervals.sort(Comparators.intervalsByStartThenEnd()); + + return nonNullSkipIntervals; + } + @VisibleForTesting + @Deprecated static List sortAndAddSkipIntervalFromLatest( Interval skipFromLatest, @Nullable List skipIntervals diff --git a/server/src/main/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfig.java index 37de2a179472..166d37540d57 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfig.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfig.java @@ -51,6 +51,7 @@ public class CatalogDataSourceCompactionConfig implements DataSourceCompactionCo @Nullable private final CompactionEngine engine; private final Period skipOffsetFromLatest; + private final Period skipOffsetFromEarliest; private final int taskPriority; @Nullable private final Map taskContext; @@ -63,6 +64,7 @@ public CatalogDataSourceCompactionConfig( @JsonProperty("dataSource") String dataSource, @JsonProperty("engine") @Nullable CompactionEngine engine, @JsonProperty("skipOffsetFromLatest") @Nullable Period skipOffsetFromLatest, + @JsonProperty("skipOffsetFromEarliest") @Nullable Period skipOffsetFromEarliest, @JsonProperty("taskPriority") @Nullable Integer taskPriority, @JsonProperty("taskContext") @Nullable Map taskContext, @JsonProperty("inputSegmentSizeBytes") @Nullable Long inputSegmentSizeBytes, @@ -72,6 +74,7 @@ public CatalogDataSourceCompactionConfig( this.dataSource = Preconditions.checkNotNull(dataSource, "dataSource"); this.engine = engine; this.skipOffsetFromLatest = skipOffsetFromLatest == null ? DEFAULT_SKIP_OFFSET_FROM_LATEST : skipOffsetFromLatest; + this.skipOffsetFromEarliest = skipOffsetFromEarliest == null ? DEFAULT_SKIP_OFFSET_FROM_EARLIEST : skipOffsetFromEarliest; this.inputSegmentSizeBytes = inputSegmentSizeBytes == null ? DEFAULT_INPUT_SEGMENT_SIZE_BYTES : inputSegmentSizeBytes; @@ -103,6 +106,13 @@ public Period getSkipOffsetFromLatest() return skipOffsetFromLatest; } + @JsonProperty + @Override + public Period getSkipOffsetFromEarliest() + { + return skipOffsetFromEarliest; + } + @JsonProperty @Override public int getTaskPriority() @@ -241,12 +251,13 @@ public boolean equals(Object o) && Objects.equals(dataSource, that.dataSource) && engine == that.engine && Objects.equals(skipOffsetFromLatest, that.skipOffsetFromLatest) + && Objects.equals(skipOffsetFromEarliest, that.skipOffsetFromEarliest) && Objects.equals(taskContext, that.taskContext); } @Override public int hashCode() { - return Objects.hash(dataSource, engine, skipOffsetFromLatest, taskPriority, taskContext, inputSegmentSizeBytes); + return Objects.hash(dataSource, engine, skipOffsetFromLatest, skipOffsetFromEarliest, taskPriority, taskContext, inputSegmentSizeBytes); } } diff --git a/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java index 4b8550ee5984..1d2325eecb9c 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java @@ -59,6 +59,7 @@ public interface DataSourceCompactionConfig // Approx. 100TB. Chosen instead of Long.MAX_VALUE to avoid overflow on web-console and other clients long DEFAULT_INPUT_SEGMENT_SIZE_BYTES = 100_000_000_000_000L; Period DEFAULT_SKIP_OFFSET_FROM_LATEST = new Period("P1D"); + Period DEFAULT_SKIP_OFFSET_FROM_EARLIEST = Period.ZERO; String getDataSource(); @@ -80,6 +81,8 @@ public interface DataSourceCompactionConfig Period getSkipOffsetFromLatest(); + Period getSkipOffsetFromEarliest(); + @Nullable UserCompactionTaskQueryTuningConfig getTuningConfig(); diff --git a/server/src/main/java/org/apache/druid/server/coordinator/InlineSchemaDataSourceCompactionConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/InlineSchemaDataSourceCompactionConfig.java index f99739d85df0..a19a68f35c7c 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/InlineSchemaDataSourceCompactionConfig.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/InlineSchemaDataSourceCompactionConfig.java @@ -56,6 +56,7 @@ public static Builder builder() @Nullable private final Integer maxRowsPerSegment; private final Period skipOffsetFromLatest; + private final Period skipOffsetFromEarliest; @Nullable private final UserCompactionTaskQueryTuningConfig tuningConfig; @Nullable @@ -84,6 +85,7 @@ public InlineSchemaDataSourceCompactionConfig( @JsonProperty("inputSegmentSizeBytes") @Nullable Long inputSegmentSizeBytes, @JsonProperty("maxRowsPerSegment") @Deprecated @Nullable Integer maxRowsPerSegment, @JsonProperty("skipOffsetFromLatest") @Nullable Period skipOffsetFromLatest, + @JsonProperty("skipOffsetFromEarliest") @Nullable Period skipOffsetFromEarliest, @JsonProperty("tuningConfig") @Nullable UserCompactionTaskQueryTuningConfig tuningConfig, @JsonProperty("granularitySpec") @Nullable UserCompactionTaskGranularityConfig granularitySpec, @JsonProperty("dimensionsSpec") @Nullable UserCompactionTaskDimensionsConfig dimensionsSpec, @@ -105,6 +107,7 @@ public InlineSchemaDataSourceCompactionConfig( : inputSegmentSizeBytes; this.maxRowsPerSegment = maxRowsPerSegment; this.skipOffsetFromLatest = skipOffsetFromLatest == null ? DEFAULT_SKIP_OFFSET_FROM_LATEST : skipOffsetFromLatest; + this.skipOffsetFromEarliest = skipOffsetFromEarliest == null ? DEFAULT_SKIP_OFFSET_FROM_EARLIEST : skipOffsetFromEarliest; this.tuningConfig = tuningConfig; this.ioConfig = ioConfig; this.granularitySpec = granularitySpec; @@ -154,6 +157,13 @@ public Period getSkipOffsetFromLatest() return skipOffsetFromLatest; } + @JsonProperty + @Override + public Period getSkipOffsetFromEarliest() + { + return skipOffsetFromEarliest; + } + @JsonProperty @Nullable @Override @@ -263,6 +273,7 @@ public boolean equals(Object o) Objects.equals(dataSource, that.dataSource) && Objects.equals(maxRowsPerSegment, that.maxRowsPerSegment) && Objects.equals(skipOffsetFromLatest, that.skipOffsetFromLatest) && + Objects.equals(skipOffsetFromEarliest, that.skipOffsetFromEarliest) && Objects.equals(tuningConfig, that.tuningConfig) && Objects.equals(granularitySpec, that.granularitySpec) && Objects.equals(dimensionsSpec, that.dimensionsSpec) && @@ -284,6 +295,7 @@ public int hashCode() inputSegmentSizeBytes, maxRowsPerSegment, skipOffsetFromLatest, + skipOffsetFromEarliest, tuningConfig, granularitySpec, dimensionsSpec, @@ -310,6 +322,7 @@ public Builder toBuilder() .withInputSegmentSizeBytes(this.inputSegmentSizeBytes) .withMaxRowsPerSegment(this.maxRowsPerSegment) .withSkipOffsetFromLatest(this.skipOffsetFromLatest) + .withSkipOffsetFromEarliest(this.skipOffsetFromEarliest) .withTuningConfig(this.tuningConfig) .withGranularitySpec(this.granularitySpec) .withDimensionsSpec(this.dimensionsSpec) @@ -329,6 +342,7 @@ public static class Builder private Long inputSegmentSizeBytes; private Integer maxRowsPerSegment; private Period skipOffsetFromLatest; + private Period skipOffsetFromEarliest; private UserCompactionTaskQueryTuningConfig tuningConfig; private UserCompactionTaskGranularityConfig granularitySpec; private UserCompactionTaskDimensionsConfig dimensionsSpec; @@ -348,6 +362,7 @@ public InlineSchemaDataSourceCompactionConfig build() inputSegmentSizeBytes, maxRowsPerSegment, skipOffsetFromLatest, + skipOffsetFromEarliest, tuningConfig, granularitySpec, dimensionsSpec, @@ -392,6 +407,12 @@ public Builder withSkipOffsetFromLatest(Period skipOffsetFromLatest) return this; } + public Builder withSkipOffsetFromEarliest(Period skipOffsetFromEarliest) + { + this.skipOffsetFromEarliest = skipOffsetFromEarliest; + return this; + } + public Builder withTuningConfig( UserCompactionTaskQueryTuningConfig tuningConfig ) diff --git a/server/src/test/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIteratorSkipOffsetTest.java b/server/src/test/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIteratorSkipOffsetTest.java new file mode 100644 index 000000000000..7c132e88ae32 --- /dev/null +++ b/server/src/test/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIteratorSkipOffsetTest.java @@ -0,0 +1,137 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.druid.server.compaction; + +import org.apache.druid.java.util.common.DateTimes; +import org.apache.druid.java.util.common.Intervals; +import org.apache.druid.java.util.common.granularity.Granularities; +import org.joda.time.DateTime; +import org.joda.time.Interval; +import org.joda.time.Period; +import org.junit.Assert; +import org.junit.Test; + +public class DataSourceCompactibleSegmentIteratorSkipOffsetTest +{ + @Test + public void test_computeEarliestSkipInterval_withoutGranularity() + { + final DateTime earliestTimestamp = DateTimes.of("2018-01-01"); + final Period skipOffset = Period.days(7); + + final Interval skipInterval = DataSourceCompactibleSegmentIterator.computeEarliestSkipInterval( + null, + earliestTimestamp, + skipOffset + ); + + // Without granularity: [2018-01-01, 2018-01-08) + Assert.assertEquals(Intervals.of("2018-01-01/2018-01-08"), skipInterval); + } + + @Test + public void test_computeEarliestSkipInterval_withGranularity() + { + final DateTime earliestTimestamp = DateTimes.of("2018-01-01"); + final Period skipOffset = Period.days(7); + + final Interval skipInterval = DataSourceCompactibleSegmentIterator.computeEarliestSkipInterval( + Granularities.DAY, + earliestTimestamp, + skipOffset + ); + + // With DAY granularity, skip offset bucket start: [2018-01-01, 2018-01-08) + Assert.assertEquals(Intervals.of("2018-01-01/2018-01-08"), skipInterval); + } + + @Test + public void test_computeEarliestSkipInterval_zeroOffset() + { + final DateTime earliestTimestamp = DateTimes.of("2018-01-01"); + final Period skipOffset = Period.ZERO; + + final Interval skipInterval = DataSourceCompactibleSegmentIterator.computeEarliestSkipInterval( + null, + earliestTimestamp, + skipOffset + ); + + // Zero offset results in an empty interval (start == end) + Assert.assertEquals(Intervals.of("2018-01-01/2018-01-01"), skipInterval); + } + + @Test + public void test_sortAndAddSkipIntervals_bothOffsets() + { + final Interval latestSkipInterval = Intervals.of("2018-12-25/2019-01-01"); + final Interval earliestSkipInterval = Intervals.of("2018-01-01/2018-01-08"); + + final java.util.List skipIntervals = DataSourceCompactibleSegmentIterator.sortAndAddSkipIntervals( + latestSkipInterval, + earliestSkipInterval, + null + ); + + // Should have both intervals, sorted + Assert.assertEquals(2, skipIntervals.size()); + Assert.assertEquals(earliestSkipInterval, skipIntervals.get(0)); + Assert.assertEquals(latestSkipInterval, skipIntervals.get(1)); + } + + @Test + public void test_sortAndAddSkipIntervals_withExistingIntervals() + { + final Interval latestSkipInterval = Intervals.of("2018-12-25/2019-01-01"); + final Interval earliestSkipInterval = Intervals.of("2018-01-01/2018-01-08"); + final java.util.List existingSkipIntervals = java.util.List.of( + Intervals.of("2018-06-01/2018-06-15") + ); + + final java.util.List skipIntervals = DataSourceCompactibleSegmentIterator.sortAndAddSkipIntervals( + latestSkipInterval, + earliestSkipInterval, + existingSkipIntervals + ); + + // Should have all three intervals, sorted + Assert.assertEquals(3, skipIntervals.size()); + Assert.assertEquals(earliestSkipInterval, skipIntervals.get(0)); + Assert.assertEquals(Intervals.of("2018-06-01/2018-06-15"), skipIntervals.get(1)); + Assert.assertEquals(latestSkipInterval, skipIntervals.get(2)); + } + + @Test + public void test_sortAndAddSkipIntervals_emptyEarliestInterval() + { + final Interval latestSkipInterval = Intervals.of("2018-12-25/2019-01-01"); + final Interval earliestSkipInterval = Intervals.of("2018-01-01/2018-01-01"); // Empty interval + + final java.util.List skipIntervals = DataSourceCompactibleSegmentIterator.sortAndAddSkipIntervals( + latestSkipInterval, + earliestSkipInterval, + null + ); + + // Empty earliest interval should not be added + Assert.assertEquals(1, skipIntervals.size()); + Assert.assertEquals(latestSkipInterval, skipIntervals.get(0)); + } +} diff --git a/server/src/test/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfigTest.java b/server/src/test/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfigTest.java index e7ed9a7d038f..83203ea5a2be 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfigTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfigTest.java @@ -116,6 +116,7 @@ public void testProjections() null, null, null, + null, METADATA_CATALOG ); @@ -135,6 +136,7 @@ public void testSerde() throws JsonProcessingException null, null, null, + null, METADATA_CATALOG ); diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java index e3b5e7eeefc3..5329931ab968 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java @@ -1071,6 +1071,7 @@ public void testCompactWithCatalogProjections() dataSource, engine, new Period("PT0H"), + null, 0, null, 500L,