| /* |
| * 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.coordinator.duty; |
| |
| import com.fasterxml.jackson.core.type.TypeReference; |
| import com.fasterxml.jackson.databind.ObjectMapper; |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.ImmutableList; |
| import com.google.common.collect.ImmutableMap; |
| import com.google.common.collect.ImmutableSet; |
| import com.google.common.collect.Iterables; |
| import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig; |
| import org.apache.druid.indexer.partitions.PartitionsSpec; |
| import org.apache.druid.jackson.DefaultObjectMapper; |
| 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.apache.druid.java.util.common.granularity.PeriodGranularity; |
| import org.apache.druid.java.util.common.guava.Comparators; |
| import org.apache.druid.segment.IndexSpec; |
| import org.apache.druid.segment.data.ConciseBitmapSerdeFactory; |
| import org.apache.druid.server.coordinator.DataSourceCompactionConfig; |
| import org.apache.druid.server.coordinator.UserCompactionTaskGranularityConfig; |
| import org.apache.druid.timeline.CompactionState; |
| import org.apache.druid.timeline.DataSegment; |
| import org.apache.druid.timeline.Partitions; |
| import org.apache.druid.timeline.VersionedIntervalTimeline; |
| import org.apache.druid.timeline.partition.NumberedShardSpec; |
| import org.apache.druid.timeline.partition.ShardSpec; |
| import org.assertj.core.api.Assertions; |
| import org.joda.time.DateTimeZone; |
| import org.joda.time.Interval; |
| import org.joda.time.Period; |
| import org.junit.Assert; |
| import org.junit.Test; |
| |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.Comparator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.TimeZone; |
| import java.util.stream.Collectors; |
| |
| public class NewestSegmentFirstPolicyTest |
| { |
| private static final String DATA_SOURCE = "dataSource"; |
| private static final long DEFAULT_SEGMENT_SIZE = 1000; |
| private static final int DEFAULT_NUM_SEGMENTS_PER_SHARD = 4; |
| |
| private final NewestSegmentFirstPolicy policy = new NewestSegmentFirstPolicy(new DefaultObjectMapper()); |
| |
| private ObjectMapper mapper = new DefaultObjectMapper(); |
| |
| @Test |
| public void testLargeOffsetAndSmallSegmentInterval() |
| { |
| final Period segmentPeriod = new Period("PT1H"); |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(10000, new Period("P2D"), null)), |
| ImmutableMap.of( |
| DATA_SOURCE, |
| createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2017-11-16T20:00:00/2017-11-17T04:00:00"), segmentPeriod), |
| new SegmentGenerateSpec(Intervals.of("2017-11-14T00:00:00/2017-11-16T07:00:00"), segmentPeriod) |
| ) |
| ), |
| Collections.emptyMap() |
| ); |
| |
| assertCompactSegmentIntervals( |
| iterator, |
| segmentPeriod, |
| Intervals.of("2017-11-14T00:00:00/2017-11-14T01:00:00"), |
| Intervals.of("2017-11-15T03:00:00/2017-11-15T04:00:00"), |
| true |
| ); |
| } |
| |
| @Test |
| public void testSmallOffsetAndLargeSegmentInterval() |
| { |
| final Period segmentPeriod = new Period("PT1H"); |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(10000, new Period("PT1M"), null)), |
| ImmutableMap.of( |
| DATA_SOURCE, |
| createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2017-11-16T20:00:00/2017-11-17T04:00:00"), segmentPeriod), |
| new SegmentGenerateSpec(Intervals.of("2017-11-14T00:00:00/2017-11-16T07:00:00"), segmentPeriod) |
| ) |
| ), |
| Collections.emptyMap() |
| ); |
| |
| assertCompactSegmentIntervals( |
| iterator, |
| segmentPeriod, |
| Intervals.of("2017-11-16T20:00:00/2017-11-16T21:00:00"), |
| Intervals.of("2017-11-17T02:00:00/2017-11-17T03:00:00"), |
| false |
| ); |
| |
| assertCompactSegmentIntervals( |
| iterator, |
| segmentPeriod, |
| Intervals.of("2017-11-14T00:00:00/2017-11-14T01:00:00"), |
| Intervals.of("2017-11-16T06:00:00/2017-11-16T07:00:00"), |
| true |
| ); |
| } |
| |
| @Test |
| public void testLargeGapInData() |
| { |
| final Period segmentPeriod = new Period("PT1H"); |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(10000, new Period("PT1H1M"), null)), |
| ImmutableMap.of( |
| DATA_SOURCE, |
| createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2017-11-16T20:00:00/2017-11-17T04:00:00"), segmentPeriod), |
| // larger gap than SegmentCompactionUtil.LOOKUP_PERIOD (1 day) |
| new SegmentGenerateSpec(Intervals.of("2017-11-14T00:00:00/2017-11-15T07:00:00"), segmentPeriod) |
| ) |
| ), |
| Collections.emptyMap() |
| ); |
| |
| assertCompactSegmentIntervals( |
| iterator, |
| segmentPeriod, |
| Intervals.of("2017-11-16T20:00:00/2017-11-16T21:00:00"), |
| Intervals.of("2017-11-17T01:00:00/2017-11-17T02:00:00"), |
| false |
| ); |
| |
| assertCompactSegmentIntervals( |
| iterator, |
| segmentPeriod, |
| Intervals.of("2017-11-14T00:00:00/2017-11-14T01:00:00"), |
| Intervals.of("2017-11-15T06:00:00/2017-11-15T07:00:00"), |
| true |
| ); |
| } |
| |
| @Test |
| public void testHugeShard() |
| { |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(10000, new Period("P1D"), null)), |
| ImmutableMap.of( |
| DATA_SOURCE, |
| createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-11-17T00:00:00/2017-11-18T03:00:00"), |
| new Period("PT1H"), |
| 200, |
| DEFAULT_NUM_SEGMENTS_PER_SHARD |
| ), |
| new SegmentGenerateSpec( |
| Intervals.of("2017-11-09T00:00:00/2017-11-17T00:00:00"), |
| new Period("P2D"), |
| 13000, // larger than target compact segment size |
| 1 |
| ), |
| new SegmentGenerateSpec( |
| Intervals.of("2017-11-05T00:00:00/2017-11-09T00:00:00"), |
| new Period("PT1H"), |
| 200, |
| DEFAULT_NUM_SEGMENTS_PER_SHARD |
| ) |
| ) |
| ), |
| Collections.emptyMap() |
| ); |
| |
| Interval lastInterval = null; |
| while (iterator.hasNext()) { |
| final List<DataSegment> segments = iterator.next(); |
| lastInterval = segments.get(0).getInterval(); |
| |
| Interval prevInterval = null; |
| for (DataSegment segment : segments) { |
| if (prevInterval != null && !prevInterval.getStart().equals(segment.getInterval().getStart())) { |
| Assert.assertEquals(prevInterval.getEnd(), segment.getInterval().getStart()); |
| } |
| |
| prevInterval = segment.getInterval(); |
| } |
| } |
| |
| Assert.assertNotNull(lastInterval); |
| Assert.assertEquals(Intervals.of("2017-11-05T00:00:00/2017-11-05T01:00:00"), lastInterval); |
| } |
| |
| @Test |
| public void testManySegmentsPerShard() |
| { |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(800000, new Period("P1D"), null)), |
| ImmutableMap.of( |
| DATA_SOURCE, |
| createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-12-04T01:00:00/2017-12-05T03:00:00"), |
| new Period("PT1H"), |
| 375, |
| 80 |
| ), |
| new SegmentGenerateSpec( |
| Intervals.of("2017-12-04T00:00:00/2017-12-04T01:00:00"), |
| new Period("PT1H"), |
| 200, |
| 150 |
| ), |
| new SegmentGenerateSpec( |
| Intervals.of("2017-12-03T18:00:00/2017-12-04T00:00:00"), |
| new Period("PT6H"), |
| 200000, |
| 1 |
| ), |
| new SegmentGenerateSpec( |
| Intervals.of("2017-12-03T11:00:00/2017-12-03T18:00:00"), |
| new Period("PT1H"), |
| 375, |
| 80 |
| ) |
| ) |
| ), |
| Collections.emptyMap() |
| ); |
| |
| Interval lastInterval = null; |
| while (iterator.hasNext()) { |
| final List<DataSegment> segments = iterator.next(); |
| lastInterval = segments.get(0).getInterval(); |
| |
| Interval prevInterval = null; |
| for (DataSegment segment : segments) { |
| if (prevInterval != null && !prevInterval.getStart().equals(segment.getInterval().getStart())) { |
| Assert.assertEquals(prevInterval.getEnd(), segment.getInterval().getStart()); |
| } |
| |
| prevInterval = segment.getInterval(); |
| } |
| } |
| |
| Assert.assertNotNull(lastInterval); |
| Assert.assertEquals(Intervals.of("2017-12-03T11:00:00/2017-12-03T12:00:00"), lastInterval); |
| } |
| |
| @Test |
| public void testSkipUnknownDataSource() |
| { |
| final String unknownDataSource = "unknown"; |
| final Period segmentPeriod = new Period("PT1H"); |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of( |
| unknownDataSource, |
| createCompactionConfig(10000, new Period("P2D"), null), |
| DATA_SOURCE, |
| createCompactionConfig(10000, new Period("P2D"), null) |
| ), |
| ImmutableMap.of( |
| DATA_SOURCE, |
| createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2017-11-16T20:00:00/2017-11-17T04:00:00"), segmentPeriod), |
| new SegmentGenerateSpec(Intervals.of("2017-11-14T00:00:00/2017-11-16T07:00:00"), segmentPeriod) |
| ) |
| ), |
| Collections.emptyMap() |
| ); |
| |
| assertCompactSegmentIntervals( |
| iterator, |
| segmentPeriod, |
| Intervals.of("2017-11-14T00:00:00/2017-11-14T01:00:00"), |
| Intervals.of("2017-11-15T03:00:00/2017-11-15T04:00:00"), |
| true |
| ); |
| } |
| |
| @Test |
| public void testClearSegmentsToCompactWhenSkippingSegments() |
| { |
| final long inputSegmentSizeBytes = 800000; |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-12-03T00:00:00/2017-12-04T00:00:00"), |
| new Period("P1D"), |
| inputSegmentSizeBytes / 2 + 10, |
| 1 |
| ), |
| new SegmentGenerateSpec( |
| Intervals.of("2017-12-02T00:00:00/2017-12-03T00:00:00"), |
| new Period("P1D"), |
| inputSegmentSizeBytes + 10, // large segment |
| 1 |
| ), |
| new SegmentGenerateSpec( |
| Intervals.of("2017-12-01T00:00:00/2017-12-02T00:00:00"), |
| new Period("P1D"), |
| inputSegmentSizeBytes / 3 + 10, |
| 2 |
| ) |
| ); |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(inputSegmentSizeBytes, new Period("P0D"), null)), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| |
| final List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-12-03/2017-12-04"), Partitions.ONLY_COMPLETE) |
| ); |
| expectedSegmentsToCompact.sort(Comparator.naturalOrder()); |
| |
| final List<DataSegment> expectedSegmentsToCompact2 = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-12-01/2017-12-02"), Partitions.ONLY_COMPLETE) |
| ); |
| expectedSegmentsToCompact2.sort(Comparator.naturalOrder()); |
| |
| Assertions.assertThat(iterator) |
| .toIterable() |
| .containsExactly(expectedSegmentsToCompact, expectedSegmentsToCompact2); |
| } |
| |
| @Test |
| public void testIfFirstSegmentIsInSkipOffset() |
| { |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-12-02T14:00:00/2017-12-03T00:00:00"), |
| new Period("PT5H"), |
| 40000, |
| 1 |
| ) |
| ); |
| |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(40000, new Period("P1D"), null)), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIfFirstSegmentOverlapsSkipOffset() |
| { |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-12-01T23:00:00/2017-12-03T00:00:00"), |
| new Period("PT5H"), |
| 40000, |
| 1 |
| ) |
| ); |
| |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(40000, new Period("P1D"), null)), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIfSegmentsSkipOffsetWithConfiguredSegmentGranularityEqual() |
| { |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2017-11-30T23:00:00/2017-12-03T00:00:00"), new Period("P1D")), |
| new SegmentGenerateSpec(Intervals.of("2017-10-14T00:00:00/2017-10-15T00:00:00"), new Period("P1D")) |
| ); |
| |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(40000, new Period("P1D"), new UserCompactionTaskGranularityConfig(Granularities.DAY, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| |
| // We should only get segments in Oct |
| final List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-10-14T00:00:00/2017-12-02T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| |
| Assert.assertTrue(iterator.hasNext()); |
| Assert.assertEquals(ImmutableSet.copyOf(expectedSegmentsToCompact), ImmutableSet.copyOf(Iterables.concat(ImmutableSet.copyOf(iterator)))); |
| } |
| |
| @Test |
| public void testIfSegmentsSkipOffsetWithConfiguredSegmentGranularityLarger() |
| { |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| // This contains segment that |
| // - Cross between month boundary of latest month (starts in Nov and ends in Dec). This should be skipped |
| // - Fully in latest month (starts in Dec and ends in Dec). This should be skipped |
| // - Does not overlap latest month (starts in Oct and ends in Oct). This should not be skipped |
| new SegmentGenerateSpec(Intervals.of("2017-11-30T23:00:00/2017-12-03T00:00:00"), new Period("PT5H")), |
| new SegmentGenerateSpec(Intervals.of("2017-10-14T00:00:00/2017-10-15T00:00:00"), new Period("PT5H")) |
| ); |
| |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(40000, new Period("P1D"), new UserCompactionTaskGranularityConfig(Granularities.MONTH, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| |
| // We should only get segments in Oct |
| final List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-10-14T00:00:00/2017-10-15T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| |
| Assert.assertTrue(iterator.hasNext()); |
| List<DataSegment> actual = iterator.next(); |
| Assert.assertEquals(expectedSegmentsToCompact.size(), actual.size()); |
| Assert.assertEquals(ImmutableSet.copyOf(expectedSegmentsToCompact), ImmutableSet.copyOf(actual)); |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIfSegmentsSkipOffsetWithConfiguredSegmentGranularitySmaller() |
| { |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2017-12-01T23:00:00/2017-12-03T00:00:00"), new Period("PT5H")), |
| new SegmentGenerateSpec(Intervals.of("2017-10-14T00:00:00/2017-10-15T00:00:00"), new Period("PT5H")) |
| ); |
| |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(40000, new Period("P1D"), new UserCompactionTaskGranularityConfig(Granularities.MINUTE, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| |
| // We should only get segments in Oct |
| final List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-10-14T00:00:00/2017-10-15T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| |
| Assert.assertTrue(iterator.hasNext()); |
| Assert.assertEquals(ImmutableSet.copyOf(expectedSegmentsToCompact), ImmutableSet.copyOf(Iterables.concat(ImmutableSet.copyOf(iterator)))); |
| } |
| |
| @Test |
| public void testWithSkipIntervals() |
| { |
| final Period segmentPeriod = new Period("PT1H"); |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(10000, new Period("P1D"), null)), |
| ImmutableMap.of( |
| DATA_SOURCE, |
| createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2017-11-16T20:00:00/2017-11-17T04:00:00"), segmentPeriod), |
| new SegmentGenerateSpec(Intervals.of("2017-11-14T00:00:00/2017-11-16T07:00:00"), segmentPeriod) |
| ) |
| ), |
| ImmutableMap.of( |
| DATA_SOURCE, |
| ImmutableList.of( |
| Intervals.of("2017-11-16T00:00:00/2017-11-17T00:00:00"), |
| Intervals.of("2017-11-15T00:00:00/2017-11-15T20:00:00"), |
| Intervals.of("2017-11-13T00:00:00/2017-11-14T01:00:00") |
| ) |
| ) |
| ); |
| |
| assertCompactSegmentIntervals( |
| iterator, |
| segmentPeriod, |
| Intervals.of("2017-11-15T20:00:00/2017-11-15T21:00:00"), |
| Intervals.of("2017-11-15T23:00:00/2017-11-16T00:00:00"), |
| false |
| ); |
| |
| assertCompactSegmentIntervals( |
| iterator, |
| segmentPeriod, |
| Intervals.of("2017-11-14T01:00:00/2017-11-14T02:00:00"), |
| Intervals.of("2017-11-14T23:00:00/2017-11-15T00:00:00"), |
| true |
| ); |
| } |
| |
| @Test |
| public void testHoleInSearchInterval() |
| { |
| final Period segmentPeriod = new Period("PT1H"); |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(10000, new Period("PT1H"), null)), |
| ImmutableMap.of( |
| DATA_SOURCE, |
| createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2017-11-16T00:00:00/2017-11-17T00:00:00"), segmentPeriod) |
| ) |
| ), |
| ImmutableMap.of( |
| DATA_SOURCE, |
| ImmutableList.of( |
| Intervals.of("2017-11-16T04:00:00/2017-11-16T10:00:00"), |
| Intervals.of("2017-11-16T14:00:00/2017-11-16T20:00:00") |
| ) |
| ) |
| ); |
| |
| assertCompactSegmentIntervals( |
| iterator, |
| segmentPeriod, |
| Intervals.of("2017-11-16T20:00:00/2017-11-16T21:00:00"), |
| Intervals.of("2017-11-16T22:00:00/2017-11-16T23:00:00"), |
| false |
| ); |
| |
| assertCompactSegmentIntervals( |
| iterator, |
| segmentPeriod, |
| Intervals.of("2017-11-16T10:00:00/2017-11-16T11:00:00"), |
| Intervals.of("2017-11-16T13:00:00/2017-11-16T14:00:00"), |
| false |
| ); |
| |
| assertCompactSegmentIntervals( |
| iterator, |
| segmentPeriod, |
| Intervals.of("2017-11-16T00:00:00/2017-11-16T01:00:00"), |
| Intervals.of("2017-11-16T03:00:00/2017-11-16T04:00:00"), |
| true |
| ); |
| } |
| |
| @Test |
| public void testIteratorReturnsSegmentsInConfiguredSegmentGranularity() |
| { |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| // Segments with day interval from Oct to Dec |
| new SegmentGenerateSpec(Intervals.of("2017-10-01T00:00:00/2017-12-31T00:00:00"), new Period("P1D")) |
| ); |
| |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(130000, new Period("P0D"), new UserCompactionTaskGranularityConfig(Granularities.MONTH, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| |
| // We should get all segments in timeline back since skip offset is P0D. |
| // However, we only need to iterator 3 times (once for each month) since the new configured segmentGranularity is MONTH. |
| // and hence iterator would return all segments bucketed to the configured segmentGranularity |
| // Month of Dec |
| Assert.assertTrue(iterator.hasNext()); |
| List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-12-01T00:00:00/2017-12-31T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertEquals( |
| ImmutableSet.copyOf(expectedSegmentsToCompact), |
| ImmutableSet.copyOf(iterator.next()) |
| ); |
| // Month of Nov |
| Assert.assertTrue(iterator.hasNext()); |
| expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-11-01T00:00:00/2017-12-01T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertEquals( |
| ImmutableSet.copyOf(expectedSegmentsToCompact), |
| ImmutableSet.copyOf(iterator.next()) |
| ); |
| // Month of Oct |
| Assert.assertTrue(iterator.hasNext()); |
| expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-10-01T00:00:00/2017-11-01T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertEquals( |
| ImmutableSet.copyOf(expectedSegmentsToCompact), |
| ImmutableSet.copyOf(iterator.next()) |
| ); |
| // No more |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIteratorReturnsSegmentsInMultipleIntervalIfConfiguredSegmentGranularityCrossBoundary() |
| { |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2020-01-01/2020-01-08"), new Period("P7D")), |
| new SegmentGenerateSpec(Intervals.of("2020-01-28/2020-02-03"), new Period("P7D")), |
| new SegmentGenerateSpec(Intervals.of("2020-02-08/2020-02-15"), new Period("P7D")) |
| ); |
| |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(130000, new Period("P0D"), new UserCompactionTaskGranularityConfig(Granularities.MONTH, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| // We should get the segment of "2020-01-28/2020-02-03" back twice when the iterator returns for Jan and when the |
| // iterator returns for Feb. |
| |
| // Month of Feb |
| List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2020-01-28/2020-02-15"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertTrue(iterator.hasNext()); |
| List<DataSegment> actual = iterator.next(); |
| Assert.assertEquals(expectedSegmentsToCompact.size(), actual.size()); |
| Assert.assertEquals(ImmutableSet.copyOf(expectedSegmentsToCompact), ImmutableSet.copyOf(actual)); |
| // Month of Jan |
| expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2020-01-01/2020-02-03"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertTrue(iterator.hasNext()); |
| actual = iterator.next(); |
| Assert.assertEquals(expectedSegmentsToCompact.size(), actual.size()); |
| Assert.assertEquals(ImmutableSet.copyOf(expectedSegmentsToCompact), ImmutableSet.copyOf(actual)); |
| // No more |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIteratorDoesNotReturnCompactedInterval() |
| { |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2017-12-01T00:00:00/2017-12-02T00:00:00"), new Period("P1D")) |
| ); |
| |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(40000, new Period("P0D"), new UserCompactionTaskGranularityConfig(Granularities.MINUTE, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| |
| final List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-12-01T00:00:00/2017-12-02T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertTrue(iterator.hasNext()); |
| Assert.assertEquals(ImmutableSet.copyOf(expectedSegmentsToCompact), ImmutableSet.copyOf(iterator.next())); |
| // Iterator should return only once since all the "minute" interval of the iterator contains the same interval |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIteratorReturnsAllMixedVersionSegmentsInConfiguredSegmentGranularity() |
| { |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2017-10-01T00:00:00/2017-10-02T00:00:00"), new Period("P1D"), "1994-04-29T00:00:00.000Z", null), |
| new SegmentGenerateSpec(Intervals.of("2017-10-01T01:00:00/2017-10-01T02:00:00"), new Period("PT1H"), "1994-04-30T00:00:00.000Z", null) |
| ); |
| |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(130000, new Period("P0D"), new UserCompactionTaskGranularityConfig(Granularities.MONTH, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| |
| // We should get all segments in timeline back since skip offset is P0D. |
| Assert.assertTrue(iterator.hasNext()); |
| List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-10-01T00:00:00/2017-10-02T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertEquals( |
| ImmutableSet.copyOf(expectedSegmentsToCompact), |
| ImmutableSet.copyOf(iterator.next()) |
| ); |
| // No more |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIteratorReturnsNothingAsSegmentsWasCompactedAndHaveSameSegmentGranularityAndSameTimezone() |
| { |
| // Same indexSpec as what is set in the auto compaction config |
| Map<String, Object> indexSpec = mapper.convertValue(new IndexSpec(), new TypeReference<Map<String, Object>>() {}); |
| // Same partitionsSpec as what is set in the auto compaction config |
| PartitionsSpec partitionsSpec = NewestSegmentFirstIterator.findPartitinosSpecFromConfig(ClientCompactionTaskQueryTuningConfig.from(null, null)); |
| |
| // Create segments that were compacted (CompactionState != null) and have segmentGranularity=DAY |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-10-01T00:00:00/2017-10-02T00:00:00"), |
| new Period("P1D"), |
| null, |
| new CompactionState(partitionsSpec, indexSpec, null) |
| ), |
| new SegmentGenerateSpec( |
| Intervals.of("2017-10-02T00:00:00/2017-10-03T00:00:00"), |
| new Period("P1D"), |
| null, |
| new CompactionState(partitionsSpec, indexSpec, null) |
| ) |
| ); |
| |
| // Auto compaction config sets segmentGranularity=DAY |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(130000, new Period("P0D"), new UserCompactionTaskGranularityConfig(Granularities.DAY, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIteratorReturnsNothingAsSegmentsWasCompactedAndHaveSameSegmentGranularityInLastCompactionState() |
| { |
| // Same indexSpec as what is set in the auto compaction config |
| Map<String, Object> indexSpec = mapper.convertValue(new IndexSpec(), new TypeReference<Map<String, Object>>() {}); |
| // Same partitionsSpec as what is set in the auto compaction config |
| PartitionsSpec partitionsSpec = NewestSegmentFirstIterator.findPartitinosSpecFromConfig(ClientCompactionTaskQueryTuningConfig.from(null, null)); |
| |
| // Create segments that were compacted (CompactionState != null) and have segmentGranularity=DAY |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-10-01T00:00:00/2017-10-02T00:00:00"), |
| new Period("P1D"), |
| null, |
| new CompactionState(partitionsSpec, indexSpec, ImmutableMap.of("segmentGranularity", "day")) |
| ), |
| new SegmentGenerateSpec( |
| Intervals.of("2017-10-02T00:00:00/2017-10-03T00:00:00"), |
| new Period("P1D"), |
| null, |
| new CompactionState(partitionsSpec, indexSpec, ImmutableMap.of("segmentGranularity", "day")) |
| ) |
| ); |
| |
| // Auto compaction config sets segmentGranularity=DAY |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(130000, new Period("P0D"), new UserCompactionTaskGranularityConfig(Granularities.DAY, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIteratorReturnsSegmentsAsSegmentsWasCompactedAndHaveDifferentSegmentGranularity() |
| { |
| // Same indexSpec as what is set in the auto compaction config |
| Map<String, Object> indexSpec = mapper.convertValue(new IndexSpec(), new TypeReference<Map<String, Object>>() {}); |
| // Same partitionsSpec as what is set in the auto compaction config |
| PartitionsSpec partitionsSpec = NewestSegmentFirstIterator.findPartitinosSpecFromConfig(ClientCompactionTaskQueryTuningConfig.from(null, null)); |
| |
| // Create segments that were compacted (CompactionState != null) and have segmentGranularity=DAY |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-10-01T00:00:00/2017-10-02T00:00:00"), |
| new Period("P1D"), |
| null, |
| new CompactionState(partitionsSpec, indexSpec, null) |
| ), |
| new SegmentGenerateSpec( |
| Intervals.of("2017-10-02T00:00:00/2017-10-03T00:00:00"), |
| new Period("P1D"), |
| null, |
| new CompactionState(partitionsSpec, indexSpec, null) |
| ) |
| ); |
| |
| // Auto compaction config sets segmentGranularity=YEAR |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(130000, new Period("P0D"), new UserCompactionTaskGranularityConfig(Granularities.YEAR, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| // We should get all segments in timeline back since skip offset is P0D. |
| Assert.assertTrue(iterator.hasNext()); |
| List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-10-01T00:00:00/2017-10-03T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertEquals( |
| ImmutableSet.copyOf(expectedSegmentsToCompact), |
| ImmutableSet.copyOf(iterator.next()) |
| ); |
| // No more |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIteratorReturnsSegmentsAsSegmentsWasCompactedAndHaveDifferentSegmentGranularityInLastCompactionState() |
| { |
| // Same indexSpec as what is set in the auto compaction config |
| Map<String, Object> indexSpec = mapper.convertValue(new IndexSpec(), new TypeReference<Map<String, Object>>() {}); |
| // Same partitionsSpec as what is set in the auto compaction config |
| PartitionsSpec partitionsSpec = NewestSegmentFirstIterator.findPartitinosSpecFromConfig(ClientCompactionTaskQueryTuningConfig.from(null, null)); |
| |
| // Create segments that were compacted (CompactionState != null) and have segmentGranularity=DAY |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-10-01T00:00:00/2017-10-02T00:00:00"), |
| new Period("P1D"), |
| null, |
| new CompactionState(partitionsSpec, indexSpec, ImmutableMap.of("segmentGranularity", "day")) |
| ), |
| new SegmentGenerateSpec( |
| Intervals.of("2017-10-02T00:00:00/2017-10-03T00:00:00"), |
| new Period("P1D"), |
| null, |
| new CompactionState(partitionsSpec, indexSpec, ImmutableMap.of("segmentGranularity", "day")) |
| ) |
| ); |
| |
| // Auto compaction config sets segmentGranularity=YEAR |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(130000, new Period("P0D"), new UserCompactionTaskGranularityConfig(Granularities.YEAR, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| // We should get all segments in timeline back since skip offset is P0D. |
| Assert.assertTrue(iterator.hasNext()); |
| List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-10-01T00:00:00/2017-10-03T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertEquals( |
| ImmutableSet.copyOf(expectedSegmentsToCompact), |
| ImmutableSet.copyOf(iterator.next()) |
| ); |
| // No more |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIteratorReturnsSegmentsAsSegmentsWasCompactedAndHaveDifferentTimezone() |
| { |
| // Same indexSpec as what is set in the auto compaction config |
| Map<String, Object> indexSpec = mapper.convertValue(new IndexSpec(), new TypeReference<Map<String, Object>>() {}); |
| // Same partitionsSpec as what is set in the auto compaction config |
| PartitionsSpec partitionsSpec = NewestSegmentFirstIterator.findPartitinosSpecFromConfig(ClientCompactionTaskQueryTuningConfig.from(null, null)); |
| |
| // Create segments that were compacted (CompactionState != null) and have segmentGranularity=DAY |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-10-02T00:00:00/2017-10-03T00:00:00"), |
| new Period("P1D"), |
| null, |
| new CompactionState(partitionsSpec, indexSpec, null) |
| ) |
| ); |
| |
| // Duration of new segmentGranularity is the same as before (P1D), |
| // but we changed the timezone from UTC to Bangkok in the auto compaction spec |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, |
| createCompactionConfig( |
| 130000, |
| new Period("P0D"), |
| new UserCompactionTaskGranularityConfig( |
| new PeriodGranularity( |
| new Period("P1D"), |
| null, |
| DateTimeZone.forTimeZone(TimeZone.getTimeZone("Asia/Bangkok")) |
| ), |
| null |
| ) |
| ) |
| ), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| // We should get all segments in timeline back since skip offset is P0D. |
| Assert.assertTrue(iterator.hasNext()); |
| List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-10-01T00:00:00/2017-10-03T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertEquals( |
| ImmutableSet.copyOf(expectedSegmentsToCompact), |
| ImmutableSet.copyOf(iterator.next()) |
| ); |
| // No more |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIteratorReturnsSegmentsAsSegmentsWasCompactedAndHaveDifferentOrigin() |
| { |
| // Same indexSpec as what is set in the auto compaction config |
| Map<String, Object> indexSpec = mapper.convertValue(new IndexSpec(), new TypeReference<Map<String, Object>>() {}); |
| // Same partitionsSpec as what is set in the auto compaction config |
| PartitionsSpec partitionsSpec = NewestSegmentFirstIterator.findPartitinosSpecFromConfig(ClientCompactionTaskQueryTuningConfig.from(null, null)); |
| |
| // Create segments that were compacted (CompactionState != null) and have segmentGranularity=DAY |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-10-02T00:00:00/2017-10-03T00:00:00"), |
| new Period("P1D"), |
| null, |
| new CompactionState(partitionsSpec, indexSpec, null) |
| ) |
| ); |
| |
| // Duration of new segmentGranularity is the same as before (P1D), but we changed the origin in the autocompaction spec |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, |
| createCompactionConfig( |
| 130000, |
| new Period("P0D"), |
| new UserCompactionTaskGranularityConfig( |
| new PeriodGranularity( |
| new Period("P1D"), |
| DateTimes.of("2012-01-02T00:05:00.000Z"), |
| DateTimeZone.UTC |
| ), |
| null |
| ) |
| ) |
| ), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| // We should get all segments in timeline back since skip offset is P0D. |
| Assert.assertTrue(iterator.hasNext()); |
| List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-10-01T00:00:00/2017-10-03T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertEquals( |
| ImmutableSet.copyOf(expectedSegmentsToCompact), |
| ImmutableSet.copyOf(iterator.next()) |
| ); |
| // No more |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIteratorReturnsSegmentsSmallerSegmentGranularityCoveringMultipleSegmentsInTimeline() |
| { |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec(Intervals.of("2017-10-01T00:00:00/2017-10-02T00:00:00"), new Period("P1D"), "1994-04-29T00:00:00.000Z", null), |
| new SegmentGenerateSpec(Intervals.of("2017-10-01T01:00:00/2017-10-01T02:00:00"), new Period("PT1H"), "1994-04-30T00:00:00.000Z", null) |
| ); |
| |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, createCompactionConfig(130000, new Period("P0D"), new UserCompactionTaskGranularityConfig(Granularities.HOUR, null))), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| |
| // We should get all segments in timeline back since skip offset is P0D. |
| // Although the first iteration only covers the last hour of 2017-10-01 (2017-10-01T23:00:00/2017-10-02T00:00:00), |
| // the iterator will returns all segment as the umbrella interval the DAY segment (2017-10-01T00:00:00/2017-10-02T00:00:00) |
| // also convers the HOUR segment (2017-10-01T01:00:00/2017-10-01T02:00:00) |
| Assert.assertTrue(iterator.hasNext()); |
| List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-10-01T00:00:00/2017-10-02T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertEquals( |
| ImmutableSet.copyOf(expectedSegmentsToCompact), |
| ImmutableSet.copyOf(iterator.next()) |
| ); |
| // No more |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| @Test |
| public void testIteratorReturnsSegmentsAsCompactionStateChangedWithCompactedStateHasSameSegmentGranularity() |
| { |
| // Different indexSpec as what is set in the auto compaction config |
| IndexSpec newIndexSpec = new IndexSpec(new ConciseBitmapSerdeFactory(), null, null, null); |
| Map<String, Object> newIndexSpecMap = mapper.convertValue(newIndexSpec, new TypeReference<Map<String, Object>>() {}); |
| PartitionsSpec partitionsSpec = NewestSegmentFirstIterator.findPartitinosSpecFromConfig(ClientCompactionTaskQueryTuningConfig.from(null, null)); |
| |
| // Create segments that were compacted (CompactionState != null) and have segmentGranularity=DAY |
| final VersionedIntervalTimeline<String, DataSegment> timeline = createTimeline( |
| new SegmentGenerateSpec( |
| Intervals.of("2017-10-02T00:00:00/2017-10-03T00:00:00"), |
| new Period("P1D"), |
| null, |
| new CompactionState(partitionsSpec, newIndexSpecMap, null) |
| ) |
| ); |
| |
| // Duration of new segmentGranularity is the same as before (P1D) |
| final CompactionSegmentIterator iterator = policy.reset( |
| ImmutableMap.of(DATA_SOURCE, |
| createCompactionConfig( |
| 130000, |
| new Period("P0D"), |
| new UserCompactionTaskGranularityConfig( |
| new PeriodGranularity( |
| new Period("P1D"), |
| null, |
| DateTimeZone.UTC |
| ), |
| null |
| ) |
| ) |
| ), |
| ImmutableMap.of(DATA_SOURCE, timeline), |
| Collections.emptyMap() |
| ); |
| // We should get all segments in timeline back since indexSpec changed |
| Assert.assertTrue(iterator.hasNext()); |
| List<DataSegment> expectedSegmentsToCompact = new ArrayList<>( |
| timeline.findNonOvershadowedObjectsInInterval(Intervals.of("2017-10-01T00:00:00/2017-10-03T00:00:00"), Partitions.ONLY_COMPLETE) |
| ); |
| Assert.assertEquals( |
| ImmutableSet.copyOf(expectedSegmentsToCompact), |
| ImmutableSet.copyOf(iterator.next()) |
| ); |
| // No more |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| |
| private static void assertCompactSegmentIntervals( |
| CompactionSegmentIterator iterator, |
| Period segmentPeriod, |
| Interval from, |
| Interval to, |
| boolean assertLast |
| ) |
| { |
| Interval expectedSegmentIntervalStart = to; |
| while (iterator.hasNext()) { |
| final List<DataSegment> segments = iterator.next(); |
| |
| final Interval firstInterval = segments.get(0).getInterval(); |
| Assert.assertTrue( |
| "Intervals should be same or abutting", |
| segments.stream().allMatch( |
| segment -> segment.getInterval().isEqual(firstInterval) || segment.getInterval().abuts(firstInterval) |
| ) |
| ); |
| |
| final List<Interval> expectedIntervals = new ArrayList<>(segments.size()); |
| for (int i = 0; i < segments.size(); i++) { |
| if (i > 0 && i % DEFAULT_NUM_SEGMENTS_PER_SHARD == 0) { |
| expectedSegmentIntervalStart = new Interval(segmentPeriod, expectedSegmentIntervalStart.getStart()); |
| } |
| expectedIntervals.add(expectedSegmentIntervalStart); |
| } |
| expectedIntervals.sort(Comparators.intervalsByStartThenEnd()); |
| |
| Assert.assertEquals( |
| expectedIntervals, |
| segments.stream().map(DataSegment::getInterval).collect(Collectors.toList()) |
| ); |
| |
| if (expectedSegmentIntervalStart.equals(from)) { |
| break; |
| } |
| expectedSegmentIntervalStart = new Interval(segmentPeriod, expectedSegmentIntervalStart.getStart()); |
| } |
| |
| if (assertLast) { |
| Assert.assertFalse(iterator.hasNext()); |
| } |
| } |
| |
| private static VersionedIntervalTimeline<String, DataSegment> createTimeline( |
| SegmentGenerateSpec... specs |
| ) |
| { |
| List<DataSegment> segments = new ArrayList<>(); |
| final String version = DateTimes.nowUtc().toString(); |
| |
| final List<SegmentGenerateSpec> orderedSpecs = Arrays.asList(specs); |
| orderedSpecs.sort(Comparator.comparing(s -> s.totalInterval, Comparators.intervalsByStartThenEnd().reversed())); |
| |
| for (SegmentGenerateSpec spec : orderedSpecs) { |
| Interval remainingInterval = spec.totalInterval; |
| |
| while (!Intervals.isEmpty(remainingInterval)) { |
| final Interval segmentInterval; |
| if (remainingInterval.toDuration().isLongerThan(spec.segmentPeriod.toStandardDuration())) { |
| segmentInterval = new Interval(spec.segmentPeriod, remainingInterval.getEnd()); |
| } else { |
| segmentInterval = remainingInterval; |
| } |
| |
| for (int i = 0; i < spec.numSegmentsPerShard; i++) { |
| final ShardSpec shardSpec = new NumberedShardSpec(i, spec.numSegmentsPerShard); |
| final DataSegment segment = new DataSegment( |
| DATA_SOURCE, |
| segmentInterval, |
| spec.version == null ? version : spec.version, |
| null, |
| ImmutableList.of(), |
| ImmutableList.of(), |
| shardSpec, |
| spec.lastCompactionState, |
| 0, |
| spec.segmentSize |
| ); |
| segments.add(segment); |
| } |
| |
| remainingInterval = SegmentCompactionUtil.removeIntervalFromEnd(remainingInterval, segmentInterval); |
| } |
| } |
| |
| return VersionedIntervalTimeline.forSegments(segments); |
| } |
| |
| private DataSourceCompactionConfig createCompactionConfig( |
| long inputSegmentSizeBytes, |
| Period skipOffsetFromLatest, |
| UserCompactionTaskGranularityConfig granularitySpec |
| ) |
| { |
| return new DataSourceCompactionConfig( |
| DATA_SOURCE, |
| 0, |
| inputSegmentSizeBytes, |
| null, |
| skipOffsetFromLatest, |
| null, |
| granularitySpec, |
| null, |
| null |
| ); |
| } |
| |
| private static class SegmentGenerateSpec |
| { |
| private final Interval totalInterval; |
| private final Period segmentPeriod; |
| private final long segmentSize; |
| private final int numSegmentsPerShard; |
| private final String version; |
| private final CompactionState lastCompactionState; |
| |
| SegmentGenerateSpec(Interval totalInterval, Period segmentPeriod) |
| { |
| this(totalInterval, segmentPeriod, null, null); |
| } |
| |
| SegmentGenerateSpec(Interval totalInterval, Period segmentPeriod, String version, CompactionState lastCompactionState) |
| { |
| this(totalInterval, segmentPeriod, DEFAULT_SEGMENT_SIZE, DEFAULT_NUM_SEGMENTS_PER_SHARD, version, lastCompactionState); |
| } |
| |
| SegmentGenerateSpec(Interval totalInterval, Period segmentPeriod, long segmentSize, int numSegmentsPerShard) |
| { |
| this(totalInterval, segmentPeriod, segmentSize, numSegmentsPerShard, null, null); |
| } |
| |
| SegmentGenerateSpec(Interval totalInterval, Period segmentPeriod, long segmentSize, int numSegmentsPerShard, String version, CompactionState lastCompactionState) |
| { |
| Preconditions.checkArgument(numSegmentsPerShard >= 1); |
| this.totalInterval = totalInterval; |
| this.segmentPeriod = segmentPeriod; |
| this.segmentSize = segmentSize; |
| this.numSegmentsPerShard = numSegmentsPerShard; |
| this.version = version; |
| this.lastCompactionState = lastCompactionState; |
| } |
| } |
| } |
