This is an automated email from the ASF dual-hosted git repository. FrankChen021 pushed a commit to branch codex/native-compaction-test-matrix in repository https://gitbox.apache.org/repos/asf/druid.git
commit 50807420f4f5d0c5a85d40fecc30bc09e77671fd Author: Frank Chen <[email protected]> AuthorDate: Tue Sep 8 18:48:21 2026 +0800 test: filter compaction parameter matrix before setup --- .../common/task/CompactionTaskRunBase.java | 257 +++++++++++--------- .../common/task/CompactionTaskRunTestCases.java | 266 +++++++++++++++++++++ .../common/task/NativeCompactionTaskRunTest.java | 90 +++---- .../druid/msq/exec/MSQCompactionTaskRunTest.java | 203 ++++++++-------- 4 files changed, 554 insertions(+), 262 deletions(-) diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java index 6dd77ff2ffe..7c890021dca 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java @@ -58,6 +58,9 @@ import org.apache.druid.indexing.common.actions.TaskActionClient; import org.apache.druid.indexing.common.actions.TaskActionTestKit; import org.apache.druid.indexing.common.config.TaskConfigBuilder; import org.apache.druid.indexing.common.task.CompactionTask.Builder; +import org.apache.druid.indexing.common.task.CompactionTaskRunTestCases.CompactionTest; +import org.apache.druid.indexing.common.task.CompactionTaskRunTestCases.Configuration; +import org.apache.druid.indexing.common.task.CompactionTaskRunTestCases.Scenario; import org.apache.druid.indexing.overlord.Segments; import org.apache.druid.java.util.common.ISE; import org.apache.druid.java.util.common.Intervals; @@ -68,6 +71,7 @@ import org.apache.druid.java.util.common.concurrent.Execs; import org.apache.druid.java.util.common.granularity.Granularities; import org.apache.druid.java.util.common.granularity.Granularity; import org.apache.druid.java.util.common.guava.Comparators; +import org.apache.druid.java.util.common.io.Closer; import org.apache.druid.query.OrderBy; import org.apache.druid.query.aggregation.AggregatorFactory; import org.apache.druid.query.aggregation.CountAggregatorFactory; @@ -122,9 +126,6 @@ import org.apache.druid.timeline.partition.PartitionIds; import org.joda.time.Interval; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Assumptions; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; import javax.annotation.Nullable; @@ -193,47 +194,42 @@ public abstract class CompactionTaskRunBase public static final TemporaryFolderExtension temporaryFolder = TemporaryFolderExtension.classScoped(); //CHECKSTYLE.ON: ConstantName - @RegisterExtension - public TaskActionTestKit taskActionTestKit = new TaskActionTestKit(); + public TaskActionTestKit taskActionTestKit; protected ObjectMapper objectMapper; protected File reportsFile; - protected final OverlordClient overlordClient; - protected final CoordinatorClient coordinatorClient; - protected final SegmentCacheManagerFactory segmentCacheManagerFactory; + protected OverlordClient overlordClient; + protected CoordinatorClient coordinatorClient; + protected SegmentCacheManagerFactory segmentCacheManagerFactory; - protected final LockGranularity lockGranularity; - protected final boolean useCentralizedDatasourceSchema; - protected final boolean useConcurrentLocks; - protected final Interval inputInterval; - protected final Granularity segmentGranularity; - protected final TestUtils testUtils; + protected LockGranularity lockGranularity; + protected boolean useCentralizedDatasourceSchema; + protected boolean useConcurrentLocks; + protected Interval inputInterval; + protected Granularity segmentGranularity; + protected TestUtils testUtils; protected ExecutorService exec; protected File localDeepStorage; - public CompactionTaskRunBase( - String name, - LockGranularity lockGranularity, - boolean useCentralizedDatasourceSchema, - boolean batchSegmentAllocation, - boolean useSegmentMetadataCache, - boolean useConcurrentLocks, - Interval inputInterval, - Granularity segmentGranularity - ) + private boolean taskActionTestKitStarted; + private boolean baseSetupStarted; + private boolean runnerSetupStarted; + + private void configure(Configuration configuration) { - this.lockGranularity = lockGranularity; - this.useCentralizedDatasourceSchema = useCentralizedDatasourceSchema; - taskActionTestKit.setUseCentralizedDatasourceSchema(useCentralizedDatasourceSchema) - .setUseSegmentMetadataCache(useSegmentMetadataCache) - .setBatchSegmentAllocation(batchSegmentAllocation); - this.useConcurrentLocks = useConcurrentLocks; - this.inputInterval = inputInterval; - this.segmentGranularity = segmentGranularity; - - reportsFile = new File(temporaryFolder.getRoot(), "reports.json"); + lockGranularity = configuration.getLockGranularity(); + useCentralizedDatasourceSchema = configuration.isUseCentralizedDatasourceSchema(); + useConcurrentLocks = configuration.isUseConcurrentLocks(); + inputInterval = configuration.getInputInterval(); + segmentGranularity = configuration.getSegmentGranularity(); + + taskActionTestKit = new TaskActionTestKit() + .setUseCentralizedDatasourceSchema(useCentralizedDatasourceSchema) + .setUseSegmentMetadataCache(configuration.isUseSegmentMetadataCache()) + .setBatchSegmentAllocation(configuration.isBatchSegmentAllocation()); + testUtils = new TestUtils(); segmentCacheManagerFactory = SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO, testUtils.getTestObjectMapper()); @@ -280,22 +276,65 @@ public abstract class CompactionTaskRunBase }; } - @BeforeEach - public void setup() throws IOException + protected final void startCase(Configuration configuration) throws Exception + { + taskActionTestKitStarted = false; + baseSetupStarted = false; + runnerSetupStarted = false; + + configure(configuration); + taskActionTestKit.before(); + taskActionTestKitStarted = true; + + baseSetupStarted = true; + setup(); + + runnerSetupStarted = true; + setUpRunner(); + } + + @AfterEach + public void cleanUpCase() throws IOException + { + try (Closer closer = Closer.create()) { + if (taskActionTestKitStarted) { + closer.register(taskActionTestKit::after); + } + if (baseSetupStarted) { + closer.register(this::teardown); + } + if (runnerSetupStarted) { + closer.register(this::tearDownRunner); + } + } + } + + protected void setup() throws IOException { exec = Execs.multiThreaded(2, "compaction-task-run-test-%d"); localDeepStorage = temporaryFolder.newFolder(); + reportsFile = new File(temporaryFolder.newFolder(), "reports.json"); } - @AfterEach - public void teardown() throws IOException + protected void teardown() + { + if (exec != null) { + exec.shutdownNow(); + } + } + + protected void setUpRunner() throws IOException { - exec.shutdownNow(); } - @Test - public void testRunWithDynamicPartitioning() throws Exception + protected void tearDownRunner() throws IOException { + } + + @CompactionTest(Scenario.RUN_WITH_DYNAMIC_PARTITIONING) + public void testRunWithDynamicPartitioning(Configuration configuration) throws Exception + { + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask = @@ -317,12 +356,10 @@ public abstract class CompactionTaskRunBase ); } - @Test - public void testRunWithHashPartitioning() throws Exception + @CompactionTest(Scenario.RUN_WITH_HASH_PARTITIONING) + public void testRunWithHashPartitioning(Configuration configuration) throws Exception { - // Hash partitioning is not supported with segment lock yet - Assumptions.assumeTrue(lockGranularity != LockGranularity.SEGMENT, "Hash partitioning is not supported with segment lock yet"); - Assumptions.assumeTrue(segmentGranularity == null, "Test null segment granularity is sufficient"); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask = @@ -360,10 +397,10 @@ public abstract class CompactionTaskRunBase } } - @Test - public void testRunCompactionTwice() throws Exception + @CompactionTest(Scenario.RUN_COMPACTION_TWICE) + public void testRunCompactionTwice(Configuration configuration) throws Exception { - Assumptions.assumeTrue(lockGranularity == LockGranularity.TIME_CHUNK); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask1 = @@ -411,10 +448,10 @@ public abstract class CompactionTaskRunBase } } - @Test - public void testRunCompactionTwiceWithSegmentLock() throws Exception + @CompactionTest(Scenario.RUN_COMPACTION_TWICE_WITH_SEGMENT_LOCK) + public void testRunCompactionTwiceWithSegmentLock(Configuration configuration) throws Exception { - Assumptions.assumeTrue(lockGranularity == LockGranularity.SEGMENT); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask1 = @@ -477,11 +514,10 @@ public abstract class CompactionTaskRunBase } } - @Test - public void testRunIndexAndCompactAtTheSameTimeForDifferentInterval() throws Exception + @CompactionTest(Scenario.RUN_INDEX_AND_COMPACT_AT_THE_SAME_TIME_FOR_DIFFERENT_INTERVAL) + public void testRunIndexAndCompactAtTheSameTimeForDifferentInterval(Configuration configuration) throws Exception { - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity) && TEST_INTERVAL.equals(inputInterval) - && lockGranularity != LockGranularity.SEGMENT, "test with defined segment granularity and interval in this test"); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask = @@ -542,10 +578,10 @@ public abstract class CompactionTaskRunBase ); } - @Test - public void testWithSegmentGranularityMisalignedInterval() throws Exception + @CompactionTest(Scenario.WITH_SEGMENT_GRANULARITY_MISALIGNED_INTERVAL) + public void testWithSegmentGranularityMisalignedInterval(Configuration configuration) throws Exception { - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity), "use Granularities.WEEK segment granularity in this test"); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); // Test when inputInterval is less than Granularities.WEEK is not allowed final CompactionTask compactionTask1 = @@ -562,10 +598,10 @@ public abstract class CompactionTaskRunBase Assertions.assertTrue(e.getMessage().contains(Granularities.WEEK.toString())); } - @Test - public void testWithSegmentGranularityMisalignedIntervalAllowed() throws Exception + @CompactionTest(Scenario.WITH_SEGMENT_GRANULARITY_MISALIGNED_INTERVAL_ALLOWED) + public void testWithSegmentGranularityMisalignedIntervalAllowed(Configuration configuration) throws Exception { - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity), "use Granularities.WEEK segment granularity in this test"); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); // Test when inputInterval is less than Granularities.WEEK is allowed final CompactionTask compactionTask1 = @@ -586,11 +622,10 @@ public abstract class CompactionTaskRunBase ); } - @Test - public void testWithSegmentGranularityMisalignedIntervalAllowed2() throws Exception + @CompactionTest(Scenario.WITH_SEGMENT_GRANULARITY_MISALIGNED_INTERVAL_ALLOWED_2) + public void testWithSegmentGranularityMisalignedIntervalAllowed2(Configuration configuration) throws Exception { - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity) && TEST_INTERVAL.equals(inputInterval) - && lockGranularity != LockGranularity.SEGMENT, "test with defined segment granularity and interval in this test"); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); // Test when inputInterval doesn't align with segment granularity final Interval interval = Intervals.of("2014-01-01T00:30:00Z/2014-01-01T01:30:00Z"); @@ -616,10 +651,10 @@ public abstract class CompactionTaskRunBase ); } - @Test - public void testCompactionWithFilterInTransformSpec() throws Exception + @CompactionTest(Scenario.COMPACTION_WITH_FILTER_IN_TRANSFORM_SPEC) + public void testCompactionWithFilterInTransformSpec(Configuration configuration) throws Exception { - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity), "test with six hour granularity is enough"); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask = compactionTaskBuilder(segmentGranularity) @@ -673,10 +708,10 @@ public abstract class CompactionTaskRunBase ); } - @Test - public void testCompactionWithNewMetricInMetricsSpec() throws Exception + @CompactionTest(Scenario.COMPACTION_WITH_NEW_METRIC_IN_METRICS_SPEC) + public void testCompactionWithNewMetricInMetricsSpec(Configuration configuration) throws Exception { - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity), "test with six hour granularity is enough"); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask = @@ -707,9 +742,10 @@ public abstract class CompactionTaskRunBase Assertions.assertEquals(expectedCompactionState, segments.get(0).getLastCompactionState()); } - @Test - public void testWithGranularitySpecNonNullQueryGranularity() throws Exception + @CompactionTest(Scenario.WITH_GRANULARITY_SPEC_NON_NULL_QUERY_GRANULARITY) + public void testWithGranularitySpecNonNullQueryGranularity(Configuration configuration) throws Exception { + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); // second queryGranularity @@ -731,10 +767,11 @@ public abstract class CompactionTaskRunBase ); } - @Test - public void testWithGranularitySpecNonNullQueryGranularityAndCoarseSegmentGranularity() throws Exception + @CompactionTest(Scenario.WITH_GRANULARITY_SPEC_NON_NULL_QUERY_GRANULARITY_AND_COARSE_SEGMENT_GRANULARITY) + public void testWithGranularitySpecNonNullQueryGranularityAndCoarseSegmentGranularity(Configuration configuration) + throws Exception { - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity) && TEST_INTERVAL.equals(inputInterval), "test with defined segment granularity and interval in this test"); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); // day segmentGranularity and day queryGranularity @@ -760,10 +797,10 @@ public abstract class CompactionTaskRunBase Assertions.assertEquals(new NumberedShardSpec(0, 1), segments.get(0).getShardSpec()); } - @Test - public void testCompactThenAppend() throws Exception + @CompactionTest(Scenario.COMPACT_THEN_APPEND) + public void testCompactThenAppend(Configuration configuration) throws Exception { - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity), "test three hour segment granularity is enough"); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask = @@ -786,13 +823,15 @@ public abstract class CompactionTaskRunBase Assertions.assertEquals(expectedSegments, usedSegments); } - @Test - public void testPartialIntervalCompactWithFinerSegmentGranularityThanFullIntervalCompactWithDropExistingTrue() + @CompactionTest( + Scenario.PARTIAL_INTERVAL_COMPACT_WITH_FINER_SEGMENT_GRANULARITY_THAN_FULL_INTERVAL_COMPACT_WITH_DROP_EXISTING_TRUE + ) + public void testPartialIntervalCompactWithFinerSegmentGranularityThanFullIntervalCompactWithDropExistingTrue( + Configuration configuration + ) throws Exception { - // This test fails with segment lock because of the bug reported in https://github.com/apache/druid/issues/10911. - Assumptions.assumeTrue(lockGranularity != LockGranularity.SEGMENT); - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity) && TEST_INTERVAL.equals(inputInterval), "test with defined segment granularity and interval in this test"); + startCase(configuration); // The following task creates (several, more than three, last time I checked, six) HOUR segments with intervals of // - 2014-01-01T00:00:00/2014-01-01T01:00:00 @@ -911,12 +950,10 @@ public abstract class CompactionTaskRunBase ); } - @Test - public void testCompactDatasourceOverIntervalWithOnlyTombstones() throws Exception + @CompactionTest(Scenario.COMPACT_DATASOURCE_OVER_INTERVAL_WITH_ONLY_TOMBSTONES) + public void testCompactDatasourceOverIntervalWithOnlyTombstones(Configuration configuration) throws Exception { - // This test fails with segment lock because of the bug reported in https://github.com/apache/druid/issues/10911. - Assumptions.assumeTrue(lockGranularity != LockGranularity.SEGMENT); - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity) && TEST_INTERVAL.equals(inputInterval), "test with defined segment granularity and interval in this test"); + startCase(configuration); // The following task creates (several, more than three, last time I checked, six) HOUR segments with intervals of // - 2014-01-01T00:00:00/2014-01-01T01:00:00 @@ -1005,13 +1042,15 @@ public abstract class CompactionTaskRunBase resultOverOnlyTombstones.rhs.getSegments().forEach(t -> Assertions.assertTrue(t.isTombstone())); } - @Test - public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullIntervalCompactWithDropExistingFalse() + @CompactionTest( + Scenario.PARTIAL_INTERVAL_COMPACT_WITH_FINER_SEGMENT_GRANULARITY_THEN_FULL_INTERVAL_COMPACT_WITH_DROP_EXISTING_FALSE + ) + public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullIntervalCompactWithDropExistingFalse( + Configuration configuration + ) throws Exception { - // This test fails with segment lock because of the bug reported in https://github.com/apache/druid/issues/10911. - Assumptions.assumeTrue(lockGranularity != LockGranularity.SEGMENT); - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity) && TEST_INTERVAL.equals(inputInterval), "test with defined segment granularity and interval in this test"); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final Set<DataSegment> expectedSegments = new HashSet<>( @@ -1059,9 +1098,10 @@ public abstract class CompactionTaskRunBase } } - @Test - public void testRunIndexAndCompactForSameSegmentAtTheSameTime() throws Exception + @CompactionTest(Scenario.RUN_INDEX_AND_COMPACT_FOR_SAME_SEGMENT_AT_THE_SAME_TIME) + public void testRunIndexAndCompactForSameSegmentAtTheSameTime(Configuration configuration) throws Exception { + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); // make sure that indexTask becomes ready first, then compactionTask becomes ready, then indexTask runs @@ -1109,9 +1149,10 @@ public abstract class CompactionTaskRunBase Assertions.assertTrue(e.getMessage().contains("not ready")); } - @Test - public void testRunIndexAndCompactForSameSegmentAtTheSameTime2() throws Exception + @CompactionTest(Scenario.RUN_INDEX_AND_COMPACT_FOR_SAME_SEGMENT_AT_THE_SAME_TIME_2) + public void testRunIndexAndCompactForSameSegmentAtTheSameTime2(Configuration configuration) throws Exception { + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask = @@ -1167,10 +1208,10 @@ public abstract class CompactionTaskRunBase Assertions.assertEquals(TaskState.FAILED, compactionResult.lhs.getStatusCode()); } - @Test - public void testRunWithSpatialDimensions() throws Exception + @CompactionTest(Scenario.RUN_WITH_SPATIAL_DIMENSIONS) + public void testRunWithSpatialDimensions(Configuration configuration) throws Exception { - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity) && TEST_INTERVAL.equals(inputInterval), "test with defined segment granularity and interval in this test"); + startCase(configuration); final List<String> spatialrows = ImmutableList.of( "2014-01-01T00:00:10Z,a,10,100,1\n", "2014-01-01T00:00:10Z,b,20,110,2\n", @@ -1276,10 +1317,10 @@ public abstract class CompactionTaskRunBase Assertions.assertEquals(spatialrows, rowsFromSegment); } - @Test - public void testRunWithAutoCastDimensions() throws Exception + @CompactionTest(Scenario.RUN_WITH_AUTO_CAST_DIMENSIONS) + public void testRunWithAutoCastDimensions(Configuration configuration) throws Exception { - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity) && TEST_INTERVAL.equals(inputInterval), "test with defined segment granularity and interval in this test"); + startCase(configuration); final List<String> rows = ImmutableList.of( "2014-01-01T00:00:10Z,a,10,100,1\n", "2014-01-01T00:00:10Z,b,20,110,2\n", @@ -1394,10 +1435,10 @@ public abstract class CompactionTaskRunBase Assertions.assertEquals(rows, rowsFromSegment); } - @Test - public void testRunWithAutoCastDimensionsSortByDimension() throws Exception + @CompactionTest(Scenario.RUN_WITH_AUTO_CAST_DIMENSIONS_SORT_BY_DIMENSION) + public void testRunWithAutoCastDimensionsSortByDimension(Configuration configuration) throws Exception { - Assumptions.assumeTrue(Granularities.SIX_HOUR.equals(segmentGranularity) && TEST_INTERVAL.equals(inputInterval), "test with defined segment granularity and interval in this test"); + startCase(configuration); // Compaction will produce one segment sorted by [x, __time], even though input rows are sorted by __time. final List<String> rows = ImmutableList.of( "2014-01-01T00:00:10Z,a,10,100,1\n", diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTestCases.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTestCases.java new file mode 100644 index 00000000000..b9cd9a850f4 --- /dev/null +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTestCases.java @@ -0,0 +1,266 @@ +/* + * 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.indexing.common.task; + +import org.apache.druid.indexing.common.LockGranularity; +import org.apache.druid.java.util.common.granularity.Granularities; +import org.apache.druid.java.util.common.granularity.Granularity; +import org.joda.time.Interval; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.ArgumentsProvider; +import org.junit.jupiter.params.provider.ArgumentsSource; + +import javax.annotation.Nullable; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Inherited; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; +import java.util.stream.Stream; + +public final class CompactionTaskRunTestCases +{ + public enum Scenario + { + RUN_WITH_DYNAMIC_PARTITIONING(Selection.ALL), + RUN_WITH_HASH_PARTITIONING(Selection.NON_SEGMENT_LOCK_WITH_NULL_GRANULARITY), + RUN_COMPACTION_TWICE(Selection.TIME_CHUNK_LOCK), + RUN_COMPACTION_TWICE_WITH_SEGMENT_LOCK(Selection.SEGMENT_LOCK), + RUN_INDEX_AND_COMPACT_AT_THE_SAME_TIME_FOR_DIFFERENT_INTERVAL( + Selection.NON_SEGMENT_LOCK_WITH_SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL + ), + WITH_SEGMENT_GRANULARITY_MISALIGNED_INTERVAL(Selection.SIX_HOUR_GRANULARITY), + WITH_SEGMENT_GRANULARITY_MISALIGNED_INTERVAL_ALLOWED(Selection.SIX_HOUR_GRANULARITY), + WITH_SEGMENT_GRANULARITY_MISALIGNED_INTERVAL_ALLOWED_2( + Selection.NON_SEGMENT_LOCK_WITH_SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL + ), + COMPACTION_WITH_FILTER_IN_TRANSFORM_SPEC(Selection.SIX_HOUR_GRANULARITY), + COMPACTION_WITH_NEW_METRIC_IN_METRICS_SPEC(Selection.SIX_HOUR_GRANULARITY), + WITH_GRANULARITY_SPEC_NON_NULL_QUERY_GRANULARITY(Selection.ALL), + WITH_GRANULARITY_SPEC_NON_NULL_QUERY_GRANULARITY_AND_COARSE_SEGMENT_GRANULARITY( + Selection.SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL + ), + COMPACT_THEN_APPEND(Selection.SIX_HOUR_GRANULARITY), + PARTIAL_INTERVAL_COMPACT_WITH_FINER_SEGMENT_GRANULARITY_THAN_FULL_INTERVAL_COMPACT_WITH_DROP_EXISTING_TRUE( + Selection.NON_SEGMENT_LOCK_WITH_SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL + ), + COMPACT_DATASOURCE_OVER_INTERVAL_WITH_ONLY_TOMBSTONES( + Selection.NON_SEGMENT_LOCK_WITH_SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL + ), + PARTIAL_INTERVAL_COMPACT_WITH_FINER_SEGMENT_GRANULARITY_THEN_FULL_INTERVAL_COMPACT_WITH_DROP_EXISTING_FALSE( + Selection.NON_SEGMENT_LOCK_WITH_SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL + ), + RUN_INDEX_AND_COMPACT_FOR_SAME_SEGMENT_AT_THE_SAME_TIME(Selection.ALL), + RUN_INDEX_AND_COMPACT_FOR_SAME_SEGMENT_AT_THE_SAME_TIME_2(Selection.ALL), + RUN_WITH_SPATIAL_DIMENSIONS(Selection.SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL), + RUN_WITH_AUTO_CAST_DIMENSIONS(Selection.SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL), + RUN_WITH_AUTO_CAST_DIMENSIONS_SORT_BY_DIMENSION(Selection.SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL); + + private final Selection selection; + + Scenario(Selection selection) + { + this.selection = selection; + } + + public boolean isApplicable(Configuration configuration) + { + return selection.isApplicable(configuration); + } + } + + private enum Selection + { + ALL, + TIME_CHUNK_LOCK, + SEGMENT_LOCK, + NON_SEGMENT_LOCK_WITH_NULL_GRANULARITY, + SIX_HOUR_GRANULARITY, + SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL, + NON_SEGMENT_LOCK_WITH_SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL; + + boolean isApplicable(Configuration configuration) + { + switch (this) { + case ALL: + return true; + case TIME_CHUNK_LOCK: + return configuration.getLockGranularity() == LockGranularity.TIME_CHUNK; + case SEGMENT_LOCK: + return configuration.getLockGranularity() == LockGranularity.SEGMENT; + case NON_SEGMENT_LOCK_WITH_NULL_GRANULARITY: + return configuration.getLockGranularity() != LockGranularity.SEGMENT + && configuration.getSegmentGranularity() == null; + case SIX_HOUR_GRANULARITY: + return Granularities.SIX_HOUR.equals(configuration.getSegmentGranularity()); + case SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL: + return Granularities.SIX_HOUR.equals(configuration.getSegmentGranularity()) + && CompactionTaskRunBase.TEST_INTERVAL.equals(configuration.getInputInterval()); + case NON_SEGMENT_LOCK_WITH_SIX_HOUR_GRANULARITY_AND_TEST_INTERVAL: + return configuration.getLockGranularity() != LockGranularity.SEGMENT + && Granularities.SIX_HOUR.equals(configuration.getSegmentGranularity()) + && CompactionTaskRunBase.TEST_INTERVAL.equals(configuration.getInputInterval()); + default: + throw new IllegalStateException("Unhandled selection " + this); + } + } + } + + public static class Configuration + { + private final LockGranularity lockGranularity; + private final boolean useCentralizedDatasourceSchema; + private final boolean batchSegmentAllocation; + private final boolean useSegmentMetadataCache; + private final boolean useConcurrentLocks; + private final Interval inputInterval; + @Nullable + private final Granularity segmentGranularity; + + public Configuration( + LockGranularity lockGranularity, + boolean useCentralizedDatasourceSchema, + boolean batchSegmentAllocation, + boolean useSegmentMetadataCache, + boolean useConcurrentLocks, + Interval inputInterval, + @Nullable Granularity segmentGranularity + ) + { + this.lockGranularity = lockGranularity; + this.useCentralizedDatasourceSchema = useCentralizedDatasourceSchema; + this.batchSegmentAllocation = batchSegmentAllocation; + this.useSegmentMetadataCache = useSegmentMetadataCache; + this.useConcurrentLocks = useConcurrentLocks; + this.inputInterval = inputInterval; + this.segmentGranularity = segmentGranularity; + } + + public LockGranularity getLockGranularity() + { + return lockGranularity; + } + + public boolean isUseCentralizedDatasourceSchema() + { + return useCentralizedDatasourceSchema; + } + + public boolean isBatchSegmentAllocation() + { + return batchSegmentAllocation; + } + + public boolean isUseSegmentMetadataCache() + { + return useSegmentMetadataCache; + } + + public boolean isUseConcurrentLocks() + { + return useConcurrentLocks; + } + + public Interval getInputInterval() + { + return inputInterval; + } + + @Nullable + public Granularity getSegmentGranularity() + { + return segmentGranularity; + } + + @Override + public String toString() + { + return "lockGranularity=" + lockGranularity + + ", useCentralizedDatasourceSchema=" + useCentralizedDatasourceSchema + + ", batchSegmentAllocation=" + batchSegmentAllocation + + ", useSegmentMetadataCache=" + useSegmentMetadataCache + + ", useConcurrentLocks=" + useConcurrentLocks + + ", inputInterval=" + inputInterval + + ", segmentGranularity=" + segmentGranularity; + } + } + + public interface ConfigurationProvider + { + Stream<Configuration> configurations(); + + default boolean isApplicable(Scenario scenario, Configuration configuration) + { + return scenario.isApplicable(configuration); + } + } + + @Inherited + @Retention(RetentionPolicy.RUNTIME) + @Target(ElementType.TYPE) + public @interface ConfigurationSource + { + Class<? extends ConfigurationProvider> value(); + } + + @Retention(RetentionPolicy.RUNTIME) + @Target(ElementType.METHOD) + @ParameterizedTest(name = "{0}") + @ArgumentsSource(ScenarioArgumentsProvider.class) + public @interface CompactionTest + { + Scenario value(); + } + + public static class ScenarioArgumentsProvider implements ArgumentsProvider + { + @Override + public Stream<? extends Arguments> provideArguments(ExtensionContext context) throws Exception + { + final CompactionTest compactionTest = context.getRequiredTestMethod().getAnnotation(CompactionTest.class); + if (compactionTest == null) { + throw new IllegalStateException("Missing @CompactionTest on " + context.getRequiredTestMethod()); + } + + final ConfigurationSource configurationSource = context.getRequiredTestClass() + .getAnnotation(ConfigurationSource.class); + if (configurationSource == null) { + throw new IllegalStateException("Missing @ConfigurationSource on " + context.getRequiredTestClass()); + } + + final ConfigurationProvider configurationProvider = configurationSource.value() + .getDeclaredConstructor() + .newInstance(); + return configurationProvider.configurations() + .filter(configuration -> configurationProvider.isApplicable( + compactionTest.value(), + configuration + )) + .map(Arguments::of); + } + } + + private CompactionTaskRunTestCases() + { + } +} diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/NativeCompactionTaskRunTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/NativeCompactionTaskRunTest.java index e9ba9d1c747..dddb2d1c14c 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/NativeCompactionTaskRunTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/NativeCompactionTaskRunTest.java @@ -21,82 +21,56 @@ package org.apache.druid.indexing.common.task; import org.apache.druid.client.indexing.ClientCompactionTaskGranularitySpec; import org.apache.druid.indexing.common.LockGranularity; -import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.indexing.common.task.CompactionTaskRunTestCases.Configuration; +import org.apache.druid.indexing.common.task.CompactionTaskRunTestCases.ConfigurationProvider; +import org.apache.druid.indexing.common.task.CompactionTaskRunTestCases.ConfigurationSource; import org.apache.druid.java.util.common.granularity.Granularities; import org.apache.druid.java.util.common.granularity.Granularity; import org.joda.time.Interval; -import org.junit.jupiter.params.ParameterizedClass; -import org.junit.jupiter.params.provider.MethodSource; import java.util.ArrayList; import java.util.List; +import java.util.stream.Stream; -@ParameterizedClass -@MethodSource("constructorFeeder") +@ConfigurationSource(NativeCompactionTaskRunTest.NativeConfigurations.class) public class NativeCompactionTaskRunTest extends CompactionTaskRunBase { - - public static Iterable<Object[]> constructorFeeder() + public static class NativeConfigurations implements ConfigurationProvider { - final List<Object[]> constructors = new ArrayList<>(); + @Override + public Stream<Configuration> configurations() + { + final List<Configuration> configurations = new ArrayList<>(); - for (LockGranularity lockGranularity : new LockGranularity[]{LockGranularity.TIME_CHUNK, LockGranularity.SEGMENT}) { - for (boolean useCentralizedDatasourceSchema : new boolean[]{true}) { - for (boolean batchSegmentAllocation : new boolean[]{false, true}) { - for (boolean useSegmentMetadataCache : new boolean[]{false, true}) { - for (boolean useConcurrentLocks : new boolean[]{false, true}) { - for (Interval inputInterval : new Interval[]{TEST_INTERVAL, TEST_INTERVAL_DAY}) { - for (Granularity segmentGran : new Granularity[]{null, Granularities.HOUR, Granularities.SIX_HOUR}) { - String name = StringUtils.format( - "lockGranularity=%s, useCentralizedDatasourceSchema=%s, batchSegmentAllocation=%s, useSegmentMetadataCache=%s, useConcurrentLocks=%s", - lockGranularity, - useCentralizedDatasourceSchema, - batchSegmentAllocation, - useSegmentMetadataCache, - useConcurrentLocks - ); - constructors.add(new Object[]{ - name, - lockGranularity, - useCentralizedDatasourceSchema, - batchSegmentAllocation, - useSegmentMetadataCache, - useConcurrentLocks, - inputInterval, - segmentGran - }); + for (final LockGranularity lockGranularity + : new LockGranularity[]{LockGranularity.TIME_CHUNK, LockGranularity.SEGMENT}) { + for (final boolean useCentralizedDatasourceSchema : new boolean[]{true}) { + for (final boolean batchSegmentAllocation : new boolean[]{false, true}) { + for (final boolean useSegmentMetadataCache : new boolean[]{false, true}) { + for (final boolean useConcurrentLocks : new boolean[]{false, true}) { + for (final Interval inputInterval : new Interval[]{TEST_INTERVAL, TEST_INTERVAL_DAY}) { + for (final Granularity segmentGranularity + : new Granularity[]{null, Granularities.HOUR, Granularities.SIX_HOUR}) { + configurations.add( + new Configuration( + lockGranularity, + useCentralizedDatasourceSchema, + batchSegmentAllocation, + useSegmentMetadataCache, + useConcurrentLocks, + inputInterval, + segmentGranularity + ) + ); + } } } } } } } - + return configurations.stream(); } - return constructors; - } - - public NativeCompactionTaskRunTest( - String name, - LockGranularity lockGranularity, - boolean useCentralizedDatasourceSchema, - boolean batchSegmentAllocation, - boolean useSegmentMetadataCache, - boolean useConcurrentLocks, - Interval inputInterval, - Granularity compactionGranularity - ) - { - super( - name, - lockGranularity, - useCentralizedDatasourceSchema, - batchSegmentAllocation, - useSegmentMetadataCache, - useConcurrentLocks, - inputInterval, - compactionGranularity - ); } @Override diff --git a/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQCompactionTaskRunTest.java b/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQCompactionTaskRunTest.java index f72cb81754b..6626a33fbfc 100644 --- a/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQCompactionTaskRunTest.java +++ b/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQCompactionTaskRunTest.java @@ -50,13 +50,17 @@ import org.apache.druid.indexing.common.actions.TaskActionClient; import org.apache.druid.indexing.common.task.CompactionIntervalSpec; import org.apache.druid.indexing.common.task.CompactionTask; import org.apache.druid.indexing.common.task.CompactionTaskRunBase; +import org.apache.druid.indexing.common.task.CompactionTaskRunTestCases.CompactionTest; +import org.apache.druid.indexing.common.task.CompactionTaskRunTestCases.Configuration; +import org.apache.druid.indexing.common.task.CompactionTaskRunTestCases.ConfigurationProvider; +import org.apache.druid.indexing.common.task.CompactionTaskRunTestCases.ConfigurationSource; +import org.apache.druid.indexing.common.task.CompactionTaskRunTestCases.Scenario; import org.apache.druid.indexing.common.task.IndexTask; import org.apache.druid.indexing.common.task.MinorCompactionInputSpec; import org.apache.druid.indexing.common.task.Tasks; import org.apache.druid.indexing.common.task.TuningConfigBuilder; import org.apache.druid.java.util.common.Intervals; import org.apache.druid.java.util.common.Pair; -import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.concurrent.Execs; import org.apache.druid.java.util.common.granularity.Granularities; import org.apache.druid.java.util.common.granularity.Granularity; @@ -105,11 +109,8 @@ import org.apache.druid.timeline.partition.NumberedShardSpec; import org.apache.druid.timeline.partition.ShardSpec; import org.joda.time.Interval; import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Assumptions; -import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Disabled; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedClass; +import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; import java.io.File; @@ -123,6 +124,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Future; import java.util.stream.Collectors; +import java.util.stream.Stream; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; @@ -132,73 +134,71 @@ import static org.mockito.Mockito.when; * Tests for CompactionTask using MSQCompactionRunner. * Extends CompactionTaskRunTest to reuse all test infrastructure. */ -@ParameterizedClass -@MethodSource("constructorFeeder") +@ConfigurationSource(MSQCompactionTaskRunTest.MsqConfigurations.class) public class MSQCompactionTaskRunTest extends CompactionTaskRunBase { private final ConcurrentHashMap<String, TaskActionClient> taskActionClients = new ConcurrentHashMap<>(); private Injector injector; - public static Iterable<Object[]> constructorFeeder() + public static class MsqConfigurations implements ConfigurationProvider { - final List<Object[]> constructors = new ArrayList<>(); - - for (LockGranularity lockGranularity : new LockGranularity[]{LockGranularity.TIME_CHUNK}) { - for (boolean useCentralizedDatasourceSchema : new boolean[]{false}) { - for (boolean batchSegmentAllocation : new boolean[]{false, true}) { - for (boolean useSegmentMetadataCache : new boolean[]{false, true}) { - for (boolean useConcurrentLocks : new boolean[]{false, true}) { - for (Interval inputInterval : new Interval[]{TEST_INTERVAL}) { - for (Granularity segmentGran : new Granularity[]{Granularities.SIX_HOUR}) { - String name = StringUtils.format( - "lockGranularity=%s, useCentralizedDatasourceSchema=%s, batchSegmentAllocation=%s, useSegmentMetadataCache=%s, useConcurrentLocks=%s", - lockGranularity, - useCentralizedDatasourceSchema, - batchSegmentAllocation, - useSegmentMetadataCache, - useConcurrentLocks - ); - constructors.add(new Object[]{ - name, - lockGranularity, - useCentralizedDatasourceSchema, - batchSegmentAllocation, - useSegmentMetadataCache, - useConcurrentLocks, - inputInterval, - segmentGran - }); + @Override + public Stream<Configuration> configurations() + { + final List<Configuration> configurations = new ArrayList<>(); + + for (final LockGranularity lockGranularity : new LockGranularity[]{LockGranularity.TIME_CHUNK}) { + for (final boolean useCentralizedDatasourceSchema : new boolean[]{false}) { + for (final boolean batchSegmentAllocation : new boolean[]{false, true}) { + for (final boolean useSegmentMetadataCache : new boolean[]{false, true}) { + for (final boolean useConcurrentLocks : new boolean[]{false, true}) { + for (final Interval inputInterval : new Interval[]{TEST_INTERVAL}) { + for (final Granularity segmentGranularity : new Granularity[]{Granularities.SIX_HOUR}) { + configurations.add( + new Configuration( + lockGranularity, + useCentralizedDatasourceSchema, + batchSegmentAllocation, + useSegmentMetadataCache, + useConcurrentLocks, + inputInterval, + segmentGranularity + ) + ); + } } } } } } } + return configurations.stream(); + } + + @Override + public boolean isApplicable(Scenario scenario, Configuration configuration) + { + if (scenario == Scenario.COMPACTION_WITH_NEW_METRIC_IN_METRICS_SPEC) { + return configuration.getSegmentGranularity() != null + && !configuration.getSegmentGranularity().isFinerThan(Granularities.SIX_HOUR); + } else if (scenario + == Scenario.PARTIAL_INTERVAL_COMPACT_WITH_FINER_SEGMENT_GRANULARITY_THAN_FULL_INTERVAL_COMPACT_WITH_DROP_EXISTING_TRUE) { + return configuration.getLockGranularity() != LockGranularity.SEGMENT + && Granularities.SIX_HOUR.equals(configuration.getSegmentGranularity()); + } + return ConfigurationProvider.super.isApplicable(scenario, configuration); } - return constructors; } - public MSQCompactionTaskRunTest( - String name, - LockGranularity lockGranularity, - boolean useCentralizedDatasourceSchema, - boolean batchSegmentAllocation, - boolean useSegmentMetadataCache, - boolean useConcurrentLocks, - Interval inputInterval, - Granularity compactionGranularities - ) + public static Stream<Configuration> concurrentLockConfigurations() { - super( - name, - lockGranularity, - useCentralizedDatasourceSchema, - batchSegmentAllocation, - useSegmentMetadataCache, - useConcurrentLocks, - inputInterval, - compactionGranularities - ); + return new MsqConfigurations().configurations().filter(Configuration::isUseConcurrentLocks); + } + + public static Stream<Configuration> timeChunkConcurrentLockConfigurations() + { + return concurrentLockConfigurations() + .filter(configuration -> configuration.getLockGranularity() == LockGranularity.TIME_CHUNK); } @Override @@ -207,8 +207,8 @@ public class MSQCompactionTaskRunTest extends CompactionTaskRunBase Preconditions.checkState(taskActionClients.put(taskId, taskActionClient) == null); } - @BeforeEach - public void setUpMSQ() + @Override + protected void setUpRunner() { objectMapper.registerModules(new MSQIndexingModule().getJacksonModules()); ComplexMetrics.registerSerde(NestedDataComplexTypeSerde.TYPE_NAME, NestedDataComplexTypeSerde.INSTANCE); @@ -295,38 +295,48 @@ public class MSQCompactionTaskRunTest extends CompactionTaskRunBase @Override @Disabled("Hash paritioning is not supported in MSQ") - @Test - public void testRunWithHashPartitioning() + @CompactionTest(Scenario.RUN_WITH_HASH_PARTITIONING) + public void testRunWithHashPartitioning(Configuration configuration) + { + } + + @Override + @Disabled("The MSQ compaction test matrix does not support segment locks") + @CompactionTest(Scenario.RUN_COMPACTION_TWICE_WITH_SEGMENT_LOCK) + public void testRunCompactionTwiceWithSegmentLock(Configuration configuration) { } @Override @Disabled("dropExisting must set to true in MSQ") - @Test - public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullIntervalCompactWithDropExistingFalse() + @CompactionTest( + Scenario.PARTIAL_INTERVAL_COMPACT_WITH_FINER_SEGMENT_GRANULARITY_THEN_FULL_INTERVAL_COMPACT_WITH_DROP_EXISTING_FALSE + ) + public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullIntervalCompactWithDropExistingFalse( + Configuration configuration + ) { } @Override @Disabled("allowNonAlignedInterval is not supported in MSQ") - @Test - public void testWithSegmentGranularityMisalignedIntervalAllowed() + @CompactionTest(Scenario.WITH_SEGMENT_GRANULARITY_MISALIGNED_INTERVAL_ALLOWED) + public void testWithSegmentGranularityMisalignedIntervalAllowed(Configuration configuration) { } @Override @Disabled("allowNonAlignedInterval is not supported in MSQ") - @Test - public void testWithSegmentGranularityMisalignedIntervalAllowed2() + @CompactionTest(Scenario.WITH_SEGMENT_GRANULARITY_MISALIGNED_INTERVAL_ALLOWED_2) + public void testWithSegmentGranularityMisalignedIntervalAllowed2(Configuration configuration) { } @Override - @Test - public void testCompactionWithNewMetricInMetricsSpec() throws Exception + @CompactionTest(Scenario.COMPACTION_WITH_NEW_METRIC_IN_METRICS_SPEC) + public void testCompactionWithNewMetricInMetricsSpec(Configuration configuration) throws Exception { - // MSQ doesn't support count aggregator - Assumptions.assumeTrue(segmentGranularity != null && !segmentGranularity.isFinerThan(Granularities.SIX_HOUR)); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask = @@ -353,17 +363,16 @@ public class MSQCompactionTaskRunTest extends CompactionTaskRunBase } @Override - @Test - public void testPartialIntervalCompactWithFinerSegmentGranularityThanFullIntervalCompactWithDropExistingTrue() + @CompactionTest( + Scenario.PARTIAL_INTERVAL_COMPACT_WITH_FINER_SEGMENT_GRANULARITY_THAN_FULL_INTERVAL_COMPACT_WITH_DROP_EXISTING_TRUE + ) + public void testPartialIntervalCompactWithFinerSegmentGranularityThanFullIntervalCompactWithDropExistingTrue( + Configuration configuration + ) throws Exception { + startCase(configuration); // This test is almost identical to base, except for fullCompactionTask, since MSQ doesn't allow disjoint intervals. - // This test fails with segment lock because of the bug reported in https://github.com/apache/druid/issues/10911. - Assumptions.assumeTrue(lockGranularity != LockGranularity.SEGMENT); - Assumptions.assumeTrue( - Granularities.SIX_HOUR.equals(segmentGranularity), - "test with defined segment granularity in this test" - ); // The following task creates (several, more than three, last time I checked, six) HOUR segments with intervals of // - 2014-01-01T00:00:00/2014-01-01T01:00:00 @@ -472,10 +481,11 @@ public class MSQCompactionTaskRunTest extends CompactionTaskRunBase ); } - @Test - public void testMSQCompactionWithConcurrentAppendCompactionLocksFirst() throws Exception + @ParameterizedTest(name = "{0}") + @MethodSource("concurrentLockConfigurations") + public void testMSQCompactionWithConcurrentAppendCompactionLocksFirst(Configuration configuration) throws Exception { - Assumptions.assumeTrue(useConcurrentLocks); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask = @@ -533,10 +543,11 @@ public class MSQCompactionTaskRunTest extends CompactionTaskRunBase } - @Test - public void testMSQCompactionWithConcurrentAppendAppendLocksFirst() throws Exception + @ParameterizedTest(name = "{0}") + @MethodSource("concurrentLockConfigurations") + public void testMSQCompactionWithConcurrentAppendAppendLocksFirst(Configuration configuration) throws Exception { - Assumptions.assumeTrue(useConcurrentLocks); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask = @@ -593,11 +604,11 @@ public class MSQCompactionTaskRunTest extends CompactionTaskRunBase verifyTaskSuccessRowsAndSchemaMatch(finalResult, 19); } - @Test - public void testMinorCompaction() throws Exception + @ParameterizedTest(name = "{0}") + @MethodSource("timeChunkConcurrentLockConfigurations") + public void testMinorCompaction(Configuration configuration) throws Exception { - Assumptions.assumeTrue(lockGranularity == LockGranularity.TIME_CHUNK); - Assumptions.assumeTrue(useConcurrentLocks, "Minor compaction depends on concurrent lock"); + startCase(configuration); verifyTaskSuccessRowsAndSchemaMatch(runIndexTask(), TOTAL_TEST_ROWS); final CompactionTask compactionTask1 = @@ -649,9 +660,11 @@ public class MSQCompactionTaskRunTest extends CompactionTaskRunBase ), usedSegments); } - @Test - public void testMinorCompactionRangePartition() throws Exception + @ParameterizedTest(name = "{0}") + @MethodSource("timeChunkConcurrentLockConfigurations") + public void testMinorCompactionRangePartition(Configuration configuration) throws Exception { + startCase(configuration); List<String> rows = ImmutableList.of( "2014-01-01T00:00:10Z,a,1\n", "2014-01-01T00:00:10Z,b,2\n", @@ -663,8 +676,6 @@ public class MSQCompactionTaskRunTest extends CompactionTaskRunBase "2014-01-01T02:00:30Z,b,2\n", "2014-01-01T02:00:30Z,c,3\n" ); - Assumptions.assumeTrue(lockGranularity == LockGranularity.TIME_CHUNK); - Assumptions.assumeTrue(useConcurrentLocks, "Minor compaction depends on concurrent lock"); verifyTaskSuccessRowsAndSchemaMatch( runTask(buildIndexTask(DEFAULT_TIMESTAMP_SPEC, DEFAULT_DIMENSIONS_SPEC, DEFAULT_INPUT_FORMAT, rows, inputInterval, false)), 9 @@ -707,11 +718,11 @@ public class MSQCompactionTaskRunTest extends CompactionTaskRunBase Assertions.assertEquals(Set.of("range"), shards.stream().map(ShardSpec::getType).collect(Collectors.toSet())); } - @Test - public void testMinorCompactionOverlappingInterval() throws Exception + @ParameterizedTest(name = "{0}") + @MethodSource("timeChunkConcurrentLockConfigurations") + public void testMinorCompactionOverlappingInterval(Configuration configuration) throws Exception { - Assumptions.assumeTrue(lockGranularity == LockGranularity.TIME_CHUNK); - Assumptions.assumeTrue(useConcurrentLocks, "Minor compaction depends on concurrent lock"); + startCase(configuration); List<String> rows = new ArrayList<>(); rows.add("2014-01-01T00:00:10Z,a1,11\n"); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
