diff --git a/.idea/inspectionProfiles/Druid.xml b/.idea/inspectionProfiles/Druid.xml
index 0889de4cc67a..88d0e34a54c7 100644
--- a/.idea/inspectionProfiles/Druid.xml
+++ b/.idea/inspectionProfiles/Druid.xml
@@ -39,9 +39,13 @@
+
+
+
+
@@ -59,6 +63,7 @@
+
-
@@ -82,7 +86,6 @@
-
@@ -111,6 +114,24 @@
+
+
@@ -154,6 +175,7 @@
+
@@ -177,105 +199,6 @@
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
@@ -301,11 +224,46 @@
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
+
@@ -313,22 +271,23 @@
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
-
-
+
+
@@ -338,16 +297,10 @@
-
-
-
-
+
-
-
-
-
+
@@ -360,10 +313,16 @@
-
+
+
+
+
+
+
+
+
+
-
-
@@ -373,40 +332,103 @@
-
+
-
-
+
+
-
+
-
-
+
-
-
-
-
+
+
+
+
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
@@ -427,6 +449,7 @@
+
@@ -481,8 +504,8 @@
-
-
+
+
@@ -493,6 +516,7 @@
+
@@ -504,6 +528,7 @@
+
\ No newline at end of file
diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTaskTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTaskTest.java
index 84ee947c8467..ab9be6408032 100644
--- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTaskTest.java
+++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTaskTest.java
@@ -87,7 +87,7 @@ public class CompactionTaskTest extends CompactionTestBase
() -> TaskBuilder
.ofTypeCompact()
.context("storeCompactionState", true)
- .ioConfig(new CompactionIntervalSpec(Intervals.of("2013-08-31/2013-09-02"), null), false);
+ .ioConfig(new CompactionIntervalSpec(Intervals.of("2013-08-31/2013-09-02"), null, null), false);
private static final Supplier PARALLEL_COMPACTION_TASK =
() -> COMPACTION_TASK.get().tuningConfig(
t -> t.withPartitionsSpec(new HashedPartitionsSpec(null, null, null))
@@ -98,7 +98,7 @@ public class CompactionTaskTest extends CompactionTestBase
() -> TaskBuilder
.ofTypeCompact()
.context("storeCompactionState", true)
- .ioConfig(new CompactionIntervalSpec(Intervals.of("2013-08-31/2013-09-02"), null), true);
+ .ioConfig(new CompactionIntervalSpec(Intervals.of("2013-08-31/2013-09-02"), null, null), true);
private static final Supplier INDEX_TASK_WITH_TIMESTAMP =
() -> MoreResources.Task.INDEX_TASK_WITH_AGGREGATORS.get().dimensions(
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionInputSpec.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionInputSpec.java
index d73a0ffe39a0..50b5f06a1cc0 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionInputSpec.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionInputSpec.java
@@ -51,4 +51,9 @@ public interface CompactionInputSpec
* @param latestSegments most recent published segments in the interval returned by {@link #findInterval}
*/
boolean validateSegments(LockGranularity lockGranularityInUse, List latestSegments);
+
+ /**
+ * Return the datasource to be used as input to the compaction task.
+ */
+ String getDataSource();
}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionIntervalSpec.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionIntervalSpec.java
index 40a4d775e031..1493e08facea 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionIntervalSpec.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionIntervalSpec.java
@@ -43,11 +43,14 @@ public class CompactionIntervalSpec implements CompactionInputSpec
private final Interval interval;
@Nullable
private final String sha256OfSortedSegmentIds;
+ @Nullable
+ private final String dataSource;
@JsonCreator
public CompactionIntervalSpec(
@JsonProperty("interval") Interval interval,
- @JsonProperty("sha256OfSortedSegmentIds") @Nullable String sha256OfSortedSegmentIds
+ @JsonProperty("sha256OfSortedSegmentIds") @Nullable String sha256OfSortedSegmentIds,
+ @JsonProperty("dataSource") @Nullable String dataSource
)
{
if (interval != null && interval.toDurationMillis() == 0) {
@@ -55,6 +58,7 @@ public CompactionIntervalSpec(
}
this.interval = interval;
this.sha256OfSortedSegmentIds = sha256OfSortedSegmentIds;
+ this.dataSource = dataSource;
}
@JsonProperty
@@ -70,6 +74,14 @@ public String getSha256OfSortedSegmentIds()
return sha256OfSortedSegmentIds;
}
+ @Override
+ @Nullable
+ @JsonProperty
+ public String getDataSource()
+ {
+ return dataSource;
+ }
+
@Override
public Interval findInterval(String dataSource)
{
@@ -97,21 +109,19 @@ public boolean validateSegments(LockGranularity lockGranularityInUse, List intervalDataSchemas = createDataSchemasForIntervals(
+ getDataSource(),
toolbox,
getTaskLockHelper().getLockGranularityToUse(),
segmentProvider,
@@ -548,6 +552,7 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
*/
@VisibleForTesting
static Map createDataSchemasForIntervals(
+ final String dataSource,
final TaskToolbox toolbox,
final LockGranularity lockGranularityInUse,
final SegmentProvider segmentProvider,
@@ -613,7 +618,7 @@ static Map createDataSchemasForIntervals(
final DataSchema dataSchema = createDataSchema(
toolbox.getEmitter(),
metricBuilder,
- segmentProvider.dataSource,
+ dataSource,
interval,
lazyFetchSegments(segmentsToCompact, toolbox.getSegmentCacheManager()),
dimensionsSpec,
@@ -633,7 +638,7 @@ static Map createDataSchemasForIntervals(
final DataSchema dataSchema = createDataSchema(
toolbox.getEmitter(),
metricBuilder,
- segmentProvider.dataSource,
+ dataSource,
JodaUtils.umbrellaInterval(
Iterables.transform(
timelineSegments,
@@ -1275,7 +1280,7 @@ public Builder(
public Builder interval(Interval interval)
{
- return inputSpec(new CompactionIntervalSpec(interval, null));
+ return inputSpec(new CompactionIntervalSpec(interval, null, null));
}
public Builder segments(List segments)
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/SpecificSegmentsSpec.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/SpecificSegmentsSpec.java
index 7b26c22a4a54..b5b77e9a8f6d 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/SpecificSegmentsSpec.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/SpecificSegmentsSpec.java
@@ -28,6 +28,7 @@
import org.apache.druid.timeline.SegmentId;
import org.joda.time.Interval;
+import javax.annotation.Nullable;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
@@ -39,18 +40,26 @@ public class SpecificSegmentsSpec implements CompactionInputSpec
private final List segments;
+ @Nullable
+ private final String dataSource;
+
public static SpecificSegmentsSpec fromSegments(List segments)
{
Preconditions.checkArgument(!segments.isEmpty(), "Empty segment list");
return new SpecificSegmentsSpec(
- segments.stream().map(segment -> segment.getId().toString()).collect(Collectors.toList())
+ segments.stream().map(segment -> segment.getId().toString()).collect(Collectors.toList()),
+ null
);
}
@JsonCreator
- public SpecificSegmentsSpec(@JsonProperty("segments") List segments)
+ public SpecificSegmentsSpec(
+ @JsonProperty("segments") List segments,
+ @JsonProperty("dataSource") @Nullable String dataSource
+ )
{
this.segments = segments;
+ this.dataSource = dataSource;
// Sort segments to use in validateSegments.
Collections.sort(this.segments);
}
@@ -61,6 +70,14 @@ public List getSegments()
return segments;
}
+ @Override
+ @Nullable
+ @JsonProperty
+ public String getDataSource()
+ {
+ return dataSource;
+ }
+
@Override
public Interval findInterval(String dataSource)
{
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java
index 1c04c7b5bd2b..ecbee4ce9b96 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java
@@ -341,7 +341,7 @@ private CompactionTask createCompactionTask(CompactionTransformSpec transformSpe
"datasource",
new SegmentCacheManagerFactory(TestIndex.INDEX_IO, MAPPER)
)
- .inputSpec(new CompactionIntervalSpec(Intervals.of("2019/2020"), "testSha256OfSortedSegmentIds"), true)
+ .inputSpec(new CompactionIntervalSpec(Intervals.of("2019/2020"), "testSha256OfSortedSegmentIds", null), true)
.tuningConfig(
TuningConfigBuilder
.forParallelIndexTask()
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionInputSpecTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionInputSpecTest.java
index 8c0db006bbfb..8814cf9af087 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionInputSpecTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionInputSpecTest.java
@@ -53,12 +53,14 @@ public static Iterable constructorFeeder()
new Object[]{
new CompactionIntervalSpec(
INTERVAL,
- SegmentUtils.hashIds(SEGMENTS)
+ SegmentUtils.hashIds(SEGMENTS),
+ null
)
},
new Object[]{
new SpecificSegmentsSpec(
- SEGMENTS.stream().map(segment -> segment.getId().toString()).collect(Collectors.toList())
+ SEGMENTS.stream().map(segment -> segment.getId().toString()).collect(Collectors.toList()),
+ null
)
}
);
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskParallelRunTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskParallelRunTest.java
index 53a7259f5bc6..cb3d45b88e72 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskParallelRunTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskParallelRunTest.java
@@ -178,7 +178,7 @@ public void testRunParallelWithDynamicPartitioningMatchCompactionState()
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
.build();
@@ -230,7 +230,7 @@ public void testRunParallelWithHashPartitioningMatchCompactionState() throws Exc
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(newTuningConfig(new HashedPartitionsSpec(null, 3, null), 2, true))
.build();
@@ -293,7 +293,7 @@ public void testRunParallelWithRangePartitioning() throws Exception
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(newTuningConfig(new SingleDimensionPartitionsSpec(7, null, "dim", false), 2, true))
.build();
@@ -342,7 +342,7 @@ public void testRunParallelWithRangePartitioningAndNoUpfrontSegmentFetching() th
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(newTuningConfig(new SingleDimensionPartitionsSpec(7, null, "dim", false), 2, true))
.dimensionsSpec(new DimensionsSpec(DimensionsSpec.getDefaultSchemas(ImmutableList.of("ts", "dim"))))
.metricsSpec(new AggregatorFactory[]{new LongSumAggregatorFactory("val", "val")})
@@ -396,7 +396,7 @@ public void testRunParallelWithMultiDimensionRangePartitioning() throws Exceptio
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(newTuningConfig(
new DimensionRangePartitionsSpec(7, null, Arrays.asList("dim1", "dim2"), false),
2,
@@ -447,7 +447,7 @@ public void testRunParallelWithRangePartitioningWithSingleTask() throws Exceptio
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(newTuningConfig(new SingleDimensionPartitionsSpec(7, null, "dim", false), 1, true))
.build();
@@ -495,7 +495,7 @@ public void testRunParallelWithMultiDimensionRangePartitioningWithSingleTask() t
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(newTuningConfig(
new DimensionRangePartitionsSpec(7, null, Arrays.asList("dim1", "dim2"), false),
1,
@@ -543,7 +543,7 @@ public void testRunCompactionStateNotStoreIfContextSetToFalse()
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
.context(ImmutableMap.of(Tasks.STORE_COMPACTION_STATE_KEY, false))
.build();
@@ -573,7 +573,7 @@ public void testRunCompactionWithFilterShouldStoreInState() throws Exception
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
.transformSpec(new CompactionTransformSpec(new SelectorDimFilter("dim", "a", null)))
.build();
@@ -624,7 +624,7 @@ public void testRunCompactionWithNewMetricsShouldStoreInState() throws Exception
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
.metricsSpec(new AggregatorFactory[]{
new CountAggregatorFactory("cnt"),
@@ -679,7 +679,7 @@ public void testCompactHashAndDynamicPartitionedSegments()
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
.build();
@@ -728,7 +728,7 @@ public void testCompactRangeAndDynamicPartitionedSegments()
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
.build();
@@ -820,7 +820,7 @@ public void testCompactionDropSegmentsOfInputIntervalIfDropFlagIsSet() throws Ex
);
final CompactionTask compactionTask = builder
// Set the dropExisting flag to true in the IOConfig of the compaction task
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null), true)
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null), true)
.tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
.granularitySpec(new ClientCompactionTaskGranularitySpec(Granularities.MINUTE, null, null))
.build();
@@ -865,7 +865,7 @@ public void testCompactionDoesNotDropSegmentsIfDropFlagNotSet() throws Exception
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
.granularitySpec(new ClientCompactionTaskGranularitySpec(Granularities.MINUTE, null, null))
.build();
@@ -906,7 +906,7 @@ public void testRunParallelWithProjections()
getSegmentCacheManagerFactory()
);
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
.build();
@@ -959,7 +959,7 @@ public void testRunParallelAddProjections()
.aggregators(new LongSumAggregatorFactory("val", "val"))
.build();
final CompactionTask compactionTask = builder
- .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+ .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
.tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
.projections(
ImmutableList.of(
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTest.java
index 7e5a15dad079..9edbf1ac06d0 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTest.java
@@ -679,7 +679,7 @@ public void testWithSegmentGranularityMisalignedInterval() throws Exception
final CompactionTask compactionTask1 = compactionTaskBuilder()
.ioConfig(
new CompactionIOConfig(
- new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null),
+ new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null, null),
false,
null
)
@@ -709,7 +709,7 @@ public void testWithSegmentGranularityMisalignedIntervalAllowed() throws Excepti
final CompactionTask compactionTask1 = compactionTaskBuilder()
.ioConfig(
new CompactionIOConfig(
- new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null),
+ new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null, null),
true,
null
)
@@ -1111,7 +1111,7 @@ public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullInterva
final CompactionTask partialCompactionTask = compactionTaskBuilder()
.segmentGranularity(Granularities.MINUTE)
// Set dropExisting to true
- .inputSpec(new CompactionIntervalSpec(compactionPartialInterval, null), true)
+ .inputSpec(new CompactionIntervalSpec(compactionPartialInterval, null, null), true)
.build();
final Pair partialCompactionResult = runTask(partialCompactionTask);
verifySchema(partialCompactionResult.rhs);
@@ -1173,7 +1173,7 @@ public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullInterva
final CompactionTask fullCompactionTask = compactionTaskBuilder()
.segmentGranularity(null)
// Set dropExisting to true
- .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null), true)
+ .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null, null), true)
.build();
// **** FULL COMPACTION ****
@@ -1263,7 +1263,7 @@ public void testCompactDatasourceOverIntervalWithOnlyTombstones() throws Excepti
final CompactionTask partialCompactionTask = compactionTaskBuilder()
.segmentGranularity(Granularities.MINUTE)
// Set dropExisting to true
- .inputSpec(new CompactionIntervalSpec(compactionPartialInterval, null), true)
+ .inputSpec(new CompactionIntervalSpec(compactionPartialInterval, null, null), true)
.build();
final Pair partialCompactionResult = runTask(partialCompactionTask);
verifySchema(partialCompactionResult.rhs);
@@ -1317,7 +1317,7 @@ public void testCompactDatasourceOverIntervalWithOnlyTombstones() throws Excepti
.segmentGranularity(null)
// Set dropExisting to true
// last 59 minutes of our 01, should be all tombstones
- .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01T01:01:00/2014-01-01T02:00:00"), null), true)
+ .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01T01:01:00/2014-01-01T02:00:00"), null, null), true)
.build();
// **** Compaction over tombstones ****
@@ -1355,7 +1355,7 @@ public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullInterva
final CompactionTask partialCompactionTask = compactionTaskBuilder()
.segmentGranularity(Granularities.MINUTE)
// Set dropExisting to false
- .inputSpec(new CompactionIntervalSpec(partialInterval, null), false)
+ .inputSpec(new CompactionIntervalSpec(partialInterval, null, null), false)
.build();
final Pair partialCompactionResult = runTask(partialCompactionTask);
@@ -1377,7 +1377,7 @@ public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullInterva
final CompactionTask fullCompactionTask = compactionTaskBuilder()
.segmentGranularity(null)
// Set dropExisting to false
- .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null), false)
+ .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null, null), false)
.build();
final Pair fullCompactionResult = runTask(fullCompactionTask);
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java
index 5d319c1013fd..f7df2d4846a4 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java
@@ -375,7 +375,7 @@ public void testCreateCompactionTaskWithGranularitySpec()
DATA_SOURCE,
segmentCacheManagerFactory
);
- builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+ builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
builder.tuningConfig(createTuningConfig());
builder.segmentGranularity(Granularities.HOUR);
final CompactionTask taskCreatedWithSegmentGranularity = builder.build();
@@ -384,7 +384,7 @@ public void testCreateCompactionTaskWithGranularitySpec()
DATA_SOURCE,
segmentCacheManagerFactory
);
- builder2.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+ builder2.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
builder2.tuningConfig(createTuningConfig());
builder2.granularitySpec(new ClientCompactionTaskGranularitySpec(Granularities.HOUR, Granularities.DAY, null));
final CompactionTask taskCreatedWithGranularitySpec = builder2.build();
@@ -401,7 +401,7 @@ public void testCreateCompactionTaskWithConflictingGranularitySpecAndSegmentGran
DATA_SOURCE,
segmentCacheManagerFactory
);
- builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+ builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
builder.tuningConfig(createTuningConfig());
builder.segmentGranularity(Granularities.HOUR);
builder.granularitySpec(new ClientCompactionTaskGranularitySpec(Granularities.MINUTE, Granularities.DAY, null));
@@ -431,7 +431,7 @@ public void testCreateCompactionTaskWithTransformSpec()
DATA_SOURCE,
segmentCacheManagerFactory
);
- builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+ builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
builder.tuningConfig(createTuningConfig());
builder.transformSpec(transformSpec);
final CompactionTask taskCreatedWithTransformSpec = builder.build();
@@ -449,7 +449,7 @@ public void testCreateCompactionTaskWithMetricsSpec()
DATA_SOURCE,
segmentCacheManagerFactory
);
- builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+ builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
builder.tuningConfig(createTuningConfig());
builder.metricsSpec(aggregatorFactories);
final CompactionTask taskCreatedWithTransformSpec = builder.build();
@@ -466,7 +466,7 @@ public void testCreateCompactionTaskWithNullSegmentGranularityInGranularitySpecA
DATA_SOURCE,
segmentCacheManagerFactory
);
- builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+ builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
builder.tuningConfig(createTuningConfig());
builder.segmentGranularity(Granularities.HOUR);
builder.granularitySpec(new ClientCompactionTaskGranularitySpec(null, Granularities.DAY, null));
@@ -494,7 +494,7 @@ public void testCreateCompactionTaskWithSameGranularitySpecAndSegmentGranularity
DATA_SOURCE,
segmentCacheManagerFactory
);
- builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+ builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
builder.tuningConfig(createTuningConfig());
builder.segmentGranularity(Granularities.HOUR);
builder.granularitySpec(new ClientCompactionTaskGranularitySpec(Granularities.HOUR, Granularities.DAY, null));
@@ -511,7 +511,7 @@ public void testSerdeWithInterval() throws IOException
);
final CompactionTask task = builder
.inputSpec(
- new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS))
+ new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null)
)
.tuningConfig(createTuningConfig())
.context(ImmutableMap.of("testKey", "testContext"))
@@ -666,7 +666,7 @@ public void testInputSourceResources()
);
final CompactionTask task = builder
.inputSpec(
- new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS))
+ new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null)
)
.tuningConfig(createTuningConfig())
.context(ImmutableMap.of("testKey", "testContext"))
@@ -770,7 +770,7 @@ public void testSegmentProviderFindSegmentsWithEmptySegmentsThrowException()
{
final SegmentProvider provider = new SegmentProvider(
"datasource",
- new CompactionIntervalSpec(Intervals.of("2021-01-01/P1D"), null)
+ new CompactionIntervalSpec(Intervals.of("2021-01-01/P1D"), null, null)
);
expectedException.expect(IllegalStateException.class);
@@ -784,9 +784,10 @@ public void testSegmentProviderFindSegmentsWithEmptySegmentsThrowException()
public void testCreateIngestionSchema() throws IOException
{
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -847,9 +848,10 @@ public void testCreateIngestionSchemaWithTargetPartitionSize() throws IOExceptio
.build();
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -911,9 +913,10 @@ public void testCreateIngestionSchemaWithMaxTotalRows() throws IOException
.build();
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -976,9 +979,10 @@ public void testCreateIngestionSchemaWithNumShards() throws IOException
.build();
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -1048,9 +1052,10 @@ public void testCreateIngestionSchemaWithCustomDimensionsSpec() throws IOExcepti
);
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
customSpec,
null,
null,
@@ -1100,9 +1105,10 @@ public void testCreateIngestionSchemaWithCustomMetricsSpec() throws IOException
};
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
customMetricsSpec,
@@ -1145,9 +1151,10 @@ public void testCreateIngestionSchemaWithCustomMetricsSpec() throws IOException
public void testCreateIngestionSchemaWithCustomSegments() throws IOException
{
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -1197,6 +1204,7 @@ public void testCreateIngestionSchemaWithDifferentSegmentSet() throws IOExceptio
// Remove one segment in the middle
segments.remove(segments.size() / 2);
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
new SegmentProvider(DATA_SOURCE, SpecificSegmentsSpec.fromSegments(segments)),
@@ -1229,6 +1237,7 @@ public void testMissingMetadata() throws IOException
indexIO.removeMetadata(Iterables.getFirst(indexIO.getQueryableIndexMap().keySet(), null));
final List segments = new ArrayList<>(SEGMENTS);
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
new SegmentProvider(DATA_SOURCE, SpecificSegmentsSpec.fromSegments(segments)),
@@ -1272,9 +1281,10 @@ public void testEmptyInterval()
public void testSegmentGranularityAndNullQueryGranularity() throws IOException
{
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -1318,9 +1328,10 @@ public void testSegmentGranularityAndNullQueryGranularity() throws IOException
public void testQueryGranularityAndNullSegmentGranularity() throws IOException
{
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -1361,9 +1372,10 @@ public void testQueryGranularityAndNullSegmentGranularity() throws IOException
public void testQueryGranularityAndSegmentGranularityNonNull() throws IOException
{
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -1414,9 +1426,10 @@ public void testQueryGranularityAndSegmentGranularityNonNull() throws IOExceptio
public void testNullGranularitySpec() throws IOException
{
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -1461,9 +1474,10 @@ public void testGranularitySpecWithNullQueryGranularityAndNullSegmentGranularity
throws IOException
{
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -1508,9 +1522,10 @@ public void testGranularitySpecWithNotNullRollup()
throws IOException
{
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -1540,9 +1555,10 @@ public void testGranularitySpecWithNullRollup()
throws IOException
{
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -1574,9 +1590,10 @@ public void testMultiValuedDimensionsProcessing()
throws IOException
{
final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+ DATA_SOURCE,
toolbox,
LockGranularity.TIME_CHUNK,
- new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+ new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
null,
null,
null,
@@ -1598,7 +1615,7 @@ public void testMSQRollupWithNoDimensionsSpecNeedsMVDInfo()
DATA_SOURCE,
segmentCacheManagerFactory
);
- builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+ builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
builder.compactionRunner(new TestMSQCompactionRunner());
final CompactionTask compactionTask = builder.build();
// granularitySpec=null should assume a possible rollup
@@ -1612,7 +1629,7 @@ public void testMSQRangePartitionWithNoDimensionsSpecNeedsMVDInfo()
DATA_SOURCE,
segmentCacheManagerFactory
);
- builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+ builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
builder.compactionRunner(new TestMSQCompactionRunner());
builder.tuningConfig(TuningConfigBuilder.forCompactionTask()
.withForceGuaranteedRollup(true)
@@ -1636,7 +1653,7 @@ public void testMSQRollupOnStringNeedsMVDInfo()
DATA_SOURCE,
segmentCacheManagerFactory
);
- builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+ builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
builder.compactionRunner(new TestMSQCompactionRunner());
builder.granularitySpec(new ClientCompactionTaskGranularitySpec(null, null, true));
@@ -1654,7 +1671,7 @@ public void testMSQRangePartitionOnStringNeedsMVDInfo()
DATA_SOURCE,
segmentCacheManagerFactory
);
- builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+ builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
builder.compactionRunner(new TestMSQCompactionRunner());
DimensionSchema stringDim = new StringDimensionSchema("string_dim_1", null, null);
diff --git a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java
index ef417a8b838f..407e02ddb4e0 100644
--- a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java
+++ b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java
@@ -688,7 +688,7 @@ private CompactionTask createCompactionTask(
context.putAll(contextParams);
builder
- .inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, null))
+ .inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null))
.tuningConfig(createTuningConfig(
indexSpec,
partitionsSpec == null ? new DynamicPartitionsSpec(100, null) : partitionsSpec