You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by fr...@apache.org on 2026/09/09 13:25:06 UTC
(druid) 01/13: test: filter compaction parameter matrix before setup
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 <fr...@outlook.com>
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: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org