This is an automated email from the ASF dual-hosted git repository.
cecemei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 1e1ee8cc454 fix: add skipIntervals in CascadingReindexingTemplate
(#20007)
1e1ee8cc454 is described below
commit 1e1ee8cc454985197de5c6ceaf31a4d55e3ca39f
Author: Cece Mei <[email protected]>
AuthorDate: Fri Aug 14 11:27:59 2026 -0700
fix: add skipIntervals in CascadingReindexingTemplate (#20007)
* skip interval and builder
* test
---
.../embedded/compact/CompactionSupervisorTest.java | 136 ++++-
.../compact/CascadingReindexingTemplate.java | 119 ++++-
.../compact/CascadingReindexingTemplateTest.java | 572 +++++++--------------
.../compact/OverlordCompactionSchedulerTest.java | 19 +-
4 files changed, 424 insertions(+), 422 deletions(-)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java
index 34ea243bbbc..96e91c82a27 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionSupervisorTest.java
@@ -98,6 +98,7 @@ import
org.apache.druid.testing.embedded.tools.JsonEventSerializer;
import org.apache.druid.testing.embedded.tools.StreamGenerator;
import
org.apache.druid.testing.embedded.tools.WikipediaStreamEventStreamGenerator;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.partition.DimensionRangeShardSpec;
import org.hamcrest.Matcher;
import org.hamcrest.MatcherAssert;
@@ -116,6 +117,7 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -420,19 +422,12 @@ public class CompactionSupervisorTest extends
EmbeddedClusterTestBase
.indexSpecRules(List.of(indexSpecRule))
.build();
- CascadingReindexingTemplate cascadingReindexingTemplate = new
CascadingReindexingTemplate(
- dataSource,
- null,
- null,
- ruleProvider,
- null,
- null,
- null,
- Granularities.HOUR,
- new DynamicPartitionsSpec(null, null),
- null,
- null
- );
+ CascadingReindexingTemplate cascadingReindexingTemplate =
CascadingReindexingTemplate.builder()
+ .forDataSource(dataSource)
+ .withRuleProvider(ruleProvider)
+ .withDefaultSegmentGranularity(Granularities.HOUR)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(null, null))
+ .build();
runCompactionWithSpec(cascadingReindexingTemplate);
waitForAllCompactionTasksToFinish();
cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator,
broker);
@@ -443,6 +438,75 @@ public class CompactionSupervisorTest extends
EmbeddedClusterTestBase
verifyEventCountOlderThan(Period.days(7), "item", "hat", 0);
}
+ @Test
+ public void
test_cascadingCompactionTemplate_skipIntervalsExcludeConfiguredWindow()
+ {
+ // Configure cluster with MSQ engine
+ final UpdateResponse updateResponse = cluster.callApi().onLeaderOverlord(
+ o -> o.updateClusterCompactionConfig(
+ new ClusterCompactionConfig(1.0, 100, null, true,
CompactionEngine.MSQ, false)
+ )
+ );
+ Assertions.assertTrue(updateResponse.isSuccess());
+
+ DateTime now = DateTimes.nowUtc();
+
+ // Both windows are older than the 1-day partitioning rule threshold
below, so the rule would
+ // normally apply to both. skipIntervals is the only thing that should
stop compactedWindow's
+ // sibling, skippedWindow, from being compacted.
+ Interval compactedWindow = new Interval(now.minusDays(31),
now.minusDays(14));
+ Interval skippedWindow = new Interval(now.minusDays(3), now.minusDays(2));
+
+ String compactedWindowEvents = generateEventsInInterval(compactedWindow,
7, Duration.ofHours(25).toMillis());
+ String skippedWindowEvents = generateEventsInInterval(skippedWindow, 5,
Duration.ofMinutes(90).toMillis());
+
+ runIngestionAtGranularity("FIFTEEN_MINUTE", compactedWindowEvents + "\n" +
skippedWindowEvents);
+
+ final Set<SegmentId> segmentIdsInSkippedWindowBeforeCompaction =
getSegmentIdsOverlapping(skippedWindow);
+
Assertions.assertFalse(segmentIdsInSkippedWindowBeforeCompaction.isEmpty());
+
+ ReindexingPartitioningRule hourRule = new ReindexingPartitioningRule(
+ "hourRule",
+ "Compact to HOUR granularity for data older than 1 day",
+ Period.days(1),
+ Granularities.HOUR,
+ new DynamicPartitionsSpec(null, null),
+ null
+ );
+
+ InlineReindexingRuleProvider ruleProvider = InlineReindexingRuleProvider
+ .builder()
+ .partitioningRules(List.of(hourRule))
+ .build();
+
+ CascadingReindexingTemplate cascadingReindexingTemplate =
CascadingReindexingTemplate.builder()
+ .forDataSource(dataSource)
+ .withRuleProvider(ruleProvider)
+ .withSkipIntervals(List.of(skippedWindow))
+ .withDefaultSegmentGranularity(Granularities.HOUR)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(null, null))
+ .build();
+ runCompactionWithSpec(cascadingReindexingTemplate);
+ waitForAllCompactionTasksToFinish();
+ cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator,
broker);
+
+ // Sanity check that the supervisor actually ran compaction on the
non-skipped window, so this
+ // test would fail if the supervisor skipped everything rather than just
the configured window.
+ Assertions.assertTrue(
+ allSegmentsOverlappingAreAlignedTo(compactedWindow,
Granularities.HOUR),
+ "expected segments in the non-skipped window to be aligned to HOUR
granularity after compaction"
+ );
+
+ // If compaction had touched skippedWindow, it would replace the original
segments with new
+ // ones (new version and/or interval). Identical segment IDs before and
after prove the
+ // configured skipIntervals window was never compacted.
+ Assertions.assertEquals(
+ segmentIdsInSkippedWindowBeforeCompaction,
+ getSegmentIdsOverlapping(skippedWindow),
+ "segments in the configured skipIntervals window should be untouched
by compaction"
+ );
+ }
+
@Test
public void
test_cascadingReindexing_withVirtualColumnOnNestedData_filtersCorrectly()
{
@@ -491,21 +555,14 @@ public class CompactionSupervisorTest extends
EmbeddedClusterTestBase
virtualColumns
);
- CascadingReindexingTemplate cascadingTemplate = new
CascadingReindexingTemplate(
- dataSource,
- null,
- null,
- InlineReindexingRuleProvider.builder()
+ CascadingReindexingTemplate cascadingTemplate =
CascadingReindexingTemplate.builder()
+ .forDataSource(dataSource)
+ .withRuleProvider(InlineReindexingRuleProvider.builder()
.deletionRules(List.of(deletionRule))
- .build(),
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(null, null),
- null,
- null
- );
+ .build())
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(null, null))
+ .build();
runCompactionWithSpec(cascadingTemplate);
@@ -1216,6 +1273,31 @@ public class CompactionSupervisorTest extends
EmbeddedClusterTestBase
.count();
}
+ private Set<SegmentId> getSegmentIdsOverlapping(Interval interval)
+ {
+ return overlord
+ .bindings()
+ .segmentsMetadataStorage()
+ .retrieveAllUsedSegments(dataSource, Segments.ONLY_VISIBLE)
+ .stream()
+ .filter(segment -> !segment.isTombstone())
+ .filter(segment -> interval.overlaps(segment.getInterval()))
+ .map(DataSegment::getId)
+ .collect(Collectors.toSet());
+ }
+
+ private boolean allSegmentsOverlappingAreAlignedTo(Interval interval,
Granularity granularity)
+ {
+ return overlord
+ .bindings()
+ .segmentsMetadataStorage()
+ .retrieveAllUsedSegments(dataSource, Segments.ONLY_VISIBLE)
+ .stream()
+ .filter(segment -> !segment.isTombstone())
+ .filter(segment -> interval.overlaps(segment.getInterval()))
+ .allMatch(segment -> granularity.isAligned(segment.getInterval()));
+ }
+
private void runIngestionAtGranularity(
String granularity,
String inlineDataCsv
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 455fced0b9d..db91eb9b5bd 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
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.annotations.VisibleForTesting;
import org.apache.druid.client.indexing.ClientCompactionRunnerInfo;
+import org.apache.druid.common.config.Configs;
import org.apache.druid.data.input.impl.AggregateProjectionSpec;
import org.apache.druid.data.input.impl.BaseTableProjectionSpec;
import org.apache.druid.error.InvalidInput;
@@ -102,6 +103,7 @@ public class CascadingReindexingTemplate implements
CompactionJobTemplate, DataS
private final long inputSegmentSizeBytes;
private final Period skipOffsetFromLatest;
private final Period skipOffsetFromNow;
+ private final List<Interval> skipIntervals;
private final Granularity defaultSegmentGranularity;
private final PartitionsSpec defaultPartitionsSpec;
@Nullable
@@ -119,6 +121,7 @@ public class CascadingReindexingTemplate implements
CompactionJobTemplate, DataS
@JsonProperty("taskContext") @Nullable Map<String, Object> taskContext,
@JsonProperty("skipOffsetFromLatest") @Nullable Period
skipOffsetFromLatest,
@JsonProperty("skipOffsetFromNow") @Nullable Period skipOffsetFromNow,
+ @JsonProperty("skipIntervals") @Nullable List<Interval> skipIntervals,
@JsonProperty("defaultSegmentGranularity") Granularity
defaultSegmentGranularity,
@JsonProperty("defaultPartitionsSpec") PartitionsSpec
defaultPartitionsSpec,
@JsonProperty("defaultPartitioningVirtualColumns") @Nullable
VirtualColumns defaultPartitioningVirtualColumns,
@@ -157,6 +160,7 @@ public class CascadingReindexingTemplate implements
CompactionJobTemplate, DataS
}
this.skipOffsetFromNow = skipOffsetFromNow;
this.skipOffsetFromLatest = skipOffsetFromLatest;
+ this.skipIntervals = Configs.valueOrDefault(skipIntervals, List.of());
this.defaultPartitioningRule = ReindexingPartitioningRule.syntheticRule(
defaultSegmentGranularity,
@@ -221,9 +225,10 @@ public class CascadingReindexingTemplate implements
CompactionJobTemplate, DataS
}
@Override
+ @JsonProperty
public List<Interval> getSkipIntervals()
{
- return List.of();
+ return skipIntervals;
}
@JsonProperty
@@ -464,6 +469,7 @@ public class CascadingReindexingTemplate implements
CompactionJobTemplate, DataS
.withInputSegmentSizeBytes(inputSegmentSizeBytes)
.withEngine(CompactionEngine.MSQ)
.withTaskContext(taskContext)
+ .withSkipIntervals(skipIntervals)
.withSkipOffsetFromLatest(Period.ZERO); // We handle skip offsets at
the timeline level, we know we want to cover the entirety of the interval
}
@@ -924,4 +930,115 @@ public class CascadingReindexingTemplate implements
CompactionJobTemplate, DataS
{
return new AggregatorFactory[0];
}
+
+ public static Builder builder()
+ {
+ return new Builder();
+ }
+
+ public static class Builder
+ {
+ private String dataSource;
+ private Integer taskPriority;
+ private Long inputSegmentSizeBytes;
+ private ReindexingRuleProvider ruleProvider;
+ private Map<String, Object> taskContext;
+ private Period skipOffsetFromLatest;
+ private Period skipOffsetFromNow;
+ private List<Interval> skipIntervals;
+ private Granularity defaultSegmentGranularity;
+ private PartitionsSpec defaultPartitionsSpec;
+ private VirtualColumns defaultPartitioningVirtualColumns;
+ private UserCompactionTaskQueryTuningConfig tuningConfig;
+
+ public CascadingReindexingTemplate build()
+ {
+ return new CascadingReindexingTemplate(
+ dataSource,
+ taskPriority,
+ inputSegmentSizeBytes,
+ ruleProvider,
+ taskContext,
+ skipOffsetFromLatest,
+ skipOffsetFromNow,
+ skipIntervals,
+ defaultSegmentGranularity,
+ defaultPartitionsSpec,
+ defaultPartitioningVirtualColumns,
+ tuningConfig
+ );
+ }
+
+ public Builder forDataSource(String dataSource)
+ {
+ this.dataSource = dataSource;
+ return this;
+ }
+
+ public Builder withTaskPriority(Integer taskPriority)
+ {
+ this.taskPriority = taskPriority;
+ return this;
+ }
+
+ public Builder withInputSegmentSizeBytes(Long inputSegmentSizeBytes)
+ {
+ this.inputSegmentSizeBytes = inputSegmentSizeBytes;
+ return this;
+ }
+
+ public Builder withRuleProvider(ReindexingRuleProvider ruleProvider)
+ {
+ this.ruleProvider = ruleProvider;
+ return this;
+ }
+
+ public Builder withTaskContext(Map<String, Object> taskContext)
+ {
+ this.taskContext = taskContext;
+ return this;
+ }
+
+ public Builder withSkipOffsetFromLatest(Period skipOffsetFromLatest)
+ {
+ this.skipOffsetFromLatest = skipOffsetFromLatest;
+ return this;
+ }
+
+ public Builder withSkipOffsetFromNow(Period skipOffsetFromNow)
+ {
+ this.skipOffsetFromNow = skipOffsetFromNow;
+ return this;
+ }
+
+ public Builder withSkipIntervals(List<Interval> skipIntervals)
+ {
+ this.skipIntervals = skipIntervals;
+ return this;
+ }
+
+ public Builder withDefaultSegmentGranularity(Granularity
defaultSegmentGranularity)
+ {
+ this.defaultSegmentGranularity = defaultSegmentGranularity;
+ return this;
+ }
+
+ public Builder withDefaultPartitionsSpec(PartitionsSpec
defaultPartitionsSpec)
+ {
+ this.defaultPartitionsSpec = defaultPartitionsSpec;
+ return this;
+ }
+
+ public Builder withDefaultPartitioningVirtualColumns(VirtualColumns
defaultPartitioningVirtualColumns)
+ {
+ this.defaultPartitioningVirtualColumns =
defaultPartitioningVirtualColumns;
+ return this;
+ }
+
+ public Builder withTuningConfig(UserCompactionTaskQueryTuningConfig
tuningConfig)
+ {
+ this.tuningConfig = tuningConfig;
+ return this;
+ }
+ }
}
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 af8d732311e..0641e1ce623 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
@@ -82,11 +82,11 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
@Test
public void test_serde() throws Exception
{
- final CascadingReindexingTemplate template = new
CascadingReindexingTemplate(
- "testDataSource",
- 50,
- 1000000L,
- InlineReindexingRuleProvider.builder()
+ final CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withTaskPriority(50)
+ .withInputSegmentSizeBytes(1000000L)
+ .withRuleProvider(InlineReindexingRuleProvider.builder()
.partitioningRules(List.of(
new ReindexingPartitioningRule(
"hourRule",
@@ -105,15 +105,11 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
null
)
))
- .build(),
- ImmutableMap.of("context_key", "context_value"),
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ .build())
+ .withTaskContext(ImmutableMap.of("context_key", "context_value"))
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
final String json = OBJECT_MAPPER.writeValueAsString(template);
final CascadingReindexingTemplate fromJson = OBJECT_MAPPER.readValue(json,
CascadingReindexingTemplate.class);
@@ -129,11 +125,11 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
@Test
public void test_serde_asDataSourceCompactionConfig() throws Exception
{
- final CascadingReindexingTemplate template = new
CascadingReindexingTemplate(
- "testDataSource",
- 30,
- 500000L,
- InlineReindexingRuleProvider.builder()
+ final CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withTaskPriority(30)
+ .withInputSegmentSizeBytes(500000L)
+ .withRuleProvider(InlineReindexingRuleProvider.builder()
.partitioningRules(List.of(
new ReindexingPartitioningRule(
"rule1",
@@ -144,15 +140,11 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
null
)
))
- .build(),
- ImmutableMap.of("key", "value"),
- null,
- null,
- Granularities.HOUR,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ .build())
+ .withTaskContext(ImmutableMap.of("key", "value"))
+ .withDefaultSegmentGranularity(Granularities.HOUR)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
// Serialize and deserialize as DataSourceCompactionConfig interface
final String json = OBJECT_MAPPER.writeValueAsString(template);
@@ -177,19 +169,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
EasyMock.expect(notReadyProvider.getType()).andReturn("mock-provider");
EasyMock.replay(notReadyProvider);
- final CascadingReindexingTemplate template = new
CascadingReindexingTemplate(
- "testDataSource",
- null,
- null,
- notReadyProvider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ final CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withRuleProvider(notReadyProvider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
// Call createCompactionJobs - should return empty list without processing
final List<CompactionJob> jobs = template.createCompactionJobs(null, null);
@@ -206,19 +191,14 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
DruidException exception = Assertions.assertThrows(
DruidException.class,
- () -> new CascadingReindexingTemplate(
- "testDataSource",
- null,
- null,
- mockProvider,
- null,
- Period.days(7), // skipOffsetFromLatest
- Period.days(3), // skipOffsetFromNow
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- )
+ () -> CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withRuleProvider(mockProvider)
+ .withSkipOffsetFromLatest(Period.days(7))
+ .withSkipOffsetFromNow(Period.days(3))
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000,
null))
+ .build()
);
Assertions.assertEquals("Cannot set both skipOffsetFromNow and
skipOffsetFromLatest", exception.getMessage());
@@ -233,19 +213,11 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
DruidException exception = Assertions.assertThrows(
DruidException.class,
- () -> new CascadingReindexingTemplate(
- null, // null dataSource
- null,
- null,
- mockProvider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- )
+ () -> CascadingReindexingTemplate.builder()
+ .withRuleProvider(mockProvider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000,
null))
+ .build()
);
Assertions.assertTrue(exception.getMessage().contains("'dataSource' cannot
be null"));
@@ -257,19 +229,11 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
{
DruidException exception = Assertions.assertThrows(
DruidException.class,
- () -> new CascadingReindexingTemplate(
- "testDataSource",
- null,
- null,
- null, // null ruleProvider
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- )
+ () -> CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000,
null))
+ .build()
);
Assertions.assertTrue(exception.getMessage().contains("'ruleProvider'
cannot be null"));
@@ -283,19 +247,11 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
DruidException exception = Assertions.assertThrows(
DruidException.class,
- () -> new CascadingReindexingTemplate(
- "testDataSource",
- null,
- null,
- mockProvider,
- null,
- null,
- null,
- null, // null defaultSegmentGranularity
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- )
+ () -> CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withRuleProvider(mockProvider)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000,
null))
+ .build()
);
Assertions.assertTrue(exception.getMessage().contains("'defaultSegmentGranularity'
cannot be null"));
@@ -314,19 +270,13 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
DruidException exception = Assertions.assertThrows(
DruidException.class,
- () -> new CascadingReindexingTemplate(
- "testDataSource",
- null,
- null,
- mockProvider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- tuningWithPartitionsSpec
- )
+ () -> CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withRuleProvider(mockProvider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000,
null))
+ .withTuningConfig(tuningWithPartitionsSpec)
+ .build()
);
Assertions.assertTrue(
@@ -549,19 +499,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
.partitioningRules(List.of(hourRule, dayRule, monthRule))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
List<IntervalPartitioningInfo> expected = List.of(
new IntervalPartitioningInfo(
@@ -639,19 +582,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
List<IntervalPartitioningInfo> expected = List.of(
new IntervalPartitioningInfo(
@@ -734,19 +670,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
// When no segment granularity rules exist, a synthetic rule is created
with the smallest period
ReindexingPartitioningRule syntheticRule =
ReindexingPartitioningRule.syntheticRule(
@@ -830,19 +759,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.HOUR,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.HOUR)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
ReindexingPartitioningRule syntheticRule =
ReindexingPartitioningRule.syntheticRule(
Granularities.HOUR,
@@ -938,19 +860,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.HOUR,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.HOUR)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
ReindexingPartitioningRule syntheticRule =
ReindexingPartitioningRule.syntheticRule(
Granularities.HOUR,
@@ -1009,19 +924,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
ReindexingRuleProvider provider =
InlineReindexingRuleProvider.builder().build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
DruidException exception = Assertions.assertThrows(
DruidException.class,
@@ -1076,19 +984,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
List<IntervalPartitioningInfo> expected = List.of(
new IntervalPartitioningInfo(
@@ -1144,19 +1045,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
List<IntervalPartitioningInfo> expected = List.of(
new IntervalPartitioningInfo(
@@ -1216,19 +1110,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
List<IntervalPartitioningInfo> expected = List.of(
new IntervalPartitioningInfo(
@@ -1280,19 +1167,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
.partitioningRules(List.of(monthRule))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
List<IntervalPartitioningInfo> expected = List.of(
new IntervalPartitioningInfo(
@@ -1347,19 +1227,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
.partitioningRules(List.of(hourRule))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
List<IntervalPartitioningInfo> expected = List.of(
new IntervalPartitioningInfo(
@@ -1439,19 +1312,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
.partitioningRules(List.of(hourRule, dayRule, monthRule))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
List<IntervalPartitioningInfo> expected = List.of(
new IntervalPartitioningInfo(
@@ -1511,19 +1377,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.MONTH, // MONTH is coarser than HOUR!
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.MONTH)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
IllegalArgumentException exception = Assertions.assertThrows(
IllegalArgumentException.class,
@@ -1574,19 +1433,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- null,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .build();
IllegalArgumentException exception = Assertions.assertThrows(
IllegalArgumentException.class,
@@ -1621,19 +1473,13 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- defaultVCs,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .withDefaultPartitioningVirtualColumns(defaultVCs)
+ .build();
ReindexingPartitioningRule syntheticRule =
ReindexingPartitioningRule.syntheticRule(
Granularities.DAY,
@@ -1681,19 +1527,13 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
))
.build();
- CascadingReindexingTemplate template = new CascadingReindexingTemplate(
- "testDS",
- null,
- null,
- provider,
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- defaultVCs,
- null
- );
+ CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDS")
+ .withRuleProvider(provider)
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .withDefaultPartitioningVirtualColumns(defaultVCs)
+ .build();
// P1M threshold: 2025-01-29 - P1M = 2024-12-29T16:15 → aligned to MONTH →
2024-12-01T00:00
ReindexingPartitioningRule syntheticRule =
ReindexingPartitioningRule.syntheticRule(
@@ -1731,26 +1571,23 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
new ExpressionVirtualColumn("vc_bucket", "timestamp_floor(__time,
'P1D')", ColumnType.LONG, TestExprMacroTable.INSTANCE)
));
- final CascadingReindexingTemplate template = new
CascadingReindexingTemplate(
- "testDataSource",
- 50,
- 1000000L,
- InlineReindexingRuleProvider.builder()
+ final CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withTaskPriority(50)
+ .withInputSegmentSizeBytes(1000000L)
+ .withRuleProvider(InlineReindexingRuleProvider.builder()
.partitioningRules(List.of(
new ReindexingPartitioningRule(
"hourRule", null, Period.days(7), Granularities.HOUR,
new DynamicPartitionsSpec(5000000, null), null
)
))
- .build(),
- ImmutableMap.of("context_key", "context_value"),
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(5000000, null),
- defaultVCs,
- null
- );
+ .build())
+ .withTaskContext(ImmutableMap.of("context_key", "context_value"))
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(5000000, null))
+ .withDefaultPartitioningVirtualColumns(defaultVCs)
+ .build();
// Need ExprMacroTable injectable for VirtualColumn deserialization
ObjectMapper mapper = new DefaultObjectMapper();
@@ -1780,19 +1617,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
@Test
public void test_validate_returnsValid_withDynamicPartitionsSpec()
{
- final CascadingReindexingTemplate template = new
CascadingReindexingTemplate(
- "testDataSource",
- null,
- null,
- InlineReindexingRuleProvider.builder().build(),
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(null, null),
- null,
- null
- );
+ final CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withRuleProvider(InlineReindexingRuleProvider.builder().build())
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(null, null))
+ .build();
CompactionConfigValidationResult result =
template.validate(CLUSTER_CONFIG);
Assertions.assertTrue(result.isValid());
@@ -1801,19 +1631,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
@Test
public void test_validate_returnsInvalid_withHashedPartitionsSpec()
{
- final CascadingReindexingTemplate template = new
CascadingReindexingTemplate(
- "testDataSource",
- null,
- null,
- InlineReindexingRuleProvider.builder().build(),
- null,
- null,
- null,
- Granularities.DAY,
- new HashedPartitionsSpec(null, 3, null),
- null,
- null
- );
+ final CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withRuleProvider(InlineReindexingRuleProvider.builder().build())
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new HashedPartitionsSpec(null, 3, null))
+ .build();
CompactionConfigValidationResult result =
template.validate(CLUSTER_CONFIG);
Assertions.assertFalse(result.isValid());
@@ -1826,19 +1649,12 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
@Test
public void test_validate_returnsInvalid_withMaxTotalRows()
{
- final CascadingReindexingTemplate template = new
CascadingReindexingTemplate(
- "testDataSource",
- null,
- null,
- InlineReindexingRuleProvider.builder().build(),
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(null, 1000L),
- null,
- null
- );
+ final CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withRuleProvider(InlineReindexingRuleProvider.builder().build())
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(null, 1000L))
+ .build();
CompactionConfigValidationResult result =
template.validate(CLUSTER_CONFIG);
Assertions.assertFalse(result.isValid());
@@ -1851,19 +1667,13 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
@Test
public void test_validate_returnsInvalid_withOneMaxNumTasks()
{
- final CascadingReindexingTemplate template = new
CascadingReindexingTemplate(
- "testDataSource",
- null,
- null,
- InlineReindexingRuleProvider.builder().build(),
- Collections.singletonMap(ClientMSQContext.CTX_MAX_NUM_TASKS, 1),
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(null, null),
- null,
- null
- );
+ final CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource("testDataSource")
+ .withRuleProvider(InlineReindexingRuleProvider.builder().build())
+
.withTaskContext(Collections.singletonMap(ClientMSQContext.CTX_MAX_NUM_TASKS,
1))
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(null, null))
+ .build();
CompactionConfigValidationResult result =
template.validate(CLUSTER_CONFIG);
Assertions.assertFalse(result.isValid());
@@ -1889,7 +1699,7 @@ public class CascadingReindexingTemplateTest extends
InitializedNullHandlingTest
)
{
super(dataSource, taskPriority, inputSegmentSizeBytes, ruleProvider,
- taskContext, skipOffsetFromLatest, skipOffsetFromNow,
Granularities.DAY,
+ taskContext, skipOffsetFromLatest, skipOffsetFromNow, null,
Granularities.DAY,
new DynamicPartitionsSpec(5000000, null), null, null
);
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java
index 9b334444124..6fa9db3acb6 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java
@@ -388,19 +388,12 @@ public class OverlordCompactionSchedulerTest
@Test
public void
test_validateCompactionConfig_delegatesToCascadingReindexingTemplate()
{
- final CascadingReindexingTemplate template = new
CascadingReindexingTemplate(
- dataSource,
- null,
- null,
- InlineReindexingRuleProvider.builder().build(),
- null,
- null,
- null,
- Granularities.DAY,
- new DynamicPartitionsSpec(null, null),
- null,
- null
- );
+ final CascadingReindexingTemplate template =
CascadingReindexingTemplate.builder()
+ .forDataSource(dataSource)
+ .withRuleProvider(InlineReindexingRuleProvider.builder().build())
+ .withDefaultSegmentGranularity(Granularities.DAY)
+ .withDefaultPartitionsSpec(new DynamicPartitionsSpec(null, null))
+ .build();
final CompactionConfigValidationResult result =
scheduler.validateCompactionConfig(template);
Assertions.assertTrue(result.isValid());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
