org.apache.druid
druid-services
${project.parent.version}
diff --git a/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/compact/CatalogCompactionTest.java b/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/compact/CatalogCompactionTest.java
index d44f3ad6593f..6c04d2eebbf2 100644
--- a/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/compact/CatalogCompactionTest.java
+++ b/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/compact/CatalogCompactionTest.java
@@ -26,6 +26,7 @@
import org.apache.druid.catalog.model.table.TableBuilder;
import org.apache.druid.catalog.sync.CatalogClient;
import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.indexing.common.task.TaskBuilder;
import org.apache.druid.indexing.compact.CompactionSupervisorSpec;
import org.apache.druid.indexing.overlord.Segments;
import org.apache.druid.java.util.common.StringUtils;
@@ -34,13 +35,11 @@
import org.apache.druid.server.coordinator.CatalogDataSourceCompactionConfig;
import org.apache.druid.server.coordinator.ClusterCompactionConfig;
import org.apache.druid.testing.embedded.EmbeddedBroker;
-import org.apache.druid.testing.embedded.EmbeddedClusterApis;
import org.apache.druid.testing.embedded.EmbeddedCoordinator;
import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
import org.apache.druid.testing.embedded.EmbeddedHistorical;
import org.apache.druid.testing.embedded.EmbeddedIndexer;
import org.apache.druid.testing.embedded.EmbeddedOverlord;
-import org.apache.druid.testing.embedded.indexing.Resources;
import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
import org.apache.druid.timeline.DataSegment;
import org.joda.time.Period;
@@ -51,7 +50,8 @@
public class CatalogCompactionTest extends EmbeddedClusterTestBase
{
- private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord()
+ .addProperty("druid.catalog.client.maxSyncRetries", "0");
private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator()
.addProperty("druid.manager.segments.useIncrementalCache", "always");
private final EmbeddedBroker broker = new EmbeddedBroker()
@@ -64,8 +64,8 @@ protected EmbeddedDruidCluster createCluster()
.useLatchableEmitter()
.addExtension(CatalogClientModule.class)
.addExtension(CatalogCoordinatorModule.class)
- .addServer(coordinator)
.addServer(overlord)
+ .addServer(coordinator)
.addServer(broker)
.addServer(new EmbeddedIndexer())
.addServer(new EmbeddedHistorical());
@@ -81,7 +81,7 @@ public void test_ingestDayGranularity_andCompactToMonthGranularity()
.segmentsMetadataStorage()
.retrieveAllUsedSegments(dataSource, Segments.ONLY_VISIBLE)
);
- Assertions.assertEquals(10, segments.size());
+ Assertions.assertEquals(3, segments.size());
segments.forEach(
segment -> Assertions.assertTrue(Granularities.DAY.isAligned(segment.getInterval()))
);
@@ -136,21 +136,26 @@ public void test_ingestDayGranularity_andCompactToMonthGranularity()
private void runIngestionAtDayGranularity()
{
final String taskId = IdUtils.getRandomId();
- final Object task = createIndexTaskForInlineData(
- taskId,
- StringUtils.replace(Resources.CSV_DATA_10_DAYS, "\n", "\\n")
- );
+ final Object task = createIndexTaskForInlineData(taskId);
cluster.callApi().onLeaderOverlord(o -> o.runTask(taskId, task));
cluster.callApi().waitForTaskToSucceed(taskId, overlord);
}
- private Object createIndexTaskForInlineData(String taskId, String inlineDataCsv)
+ private Object createIndexTaskForInlineData(String taskId)
{
- return EmbeddedClusterApis.createTaskFromPayload(
- taskId,
- StringUtils.format(Resources.INDEX_TASK_PAYLOAD_WITH_INLINE_DATA, inlineDataCsv, dataSource)
- );
+ final String inlineDataCsv =
+ "2025-06-01T00:00:00.000Z,shirt,105"
+ + "\n2025-06-02T00:00:00.000Z,trousers,210"
+ + "\n2025-06-03T00:00:00.000Z,jeans,150";
+ return TaskBuilder.ofTypeIndex()
+ .dataSource(dataSource)
+ .isoTimestampColumn("time")
+ .csvInputFormatWithColumns("time", "item", "value")
+ .inlineInputSourceWithData(inlineDataCsv)
+ .segmentGranularity("DAY")
+ .dimensions()
+ .withId(taskId);
}
private void enableCompactionSupervisor()
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskBuilder.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskBuilder.java
new file mode 100644
index 000000000000..e616471034c0
--- /dev/null
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskBuilder.java
@@ -0,0 +1,455 @@
+/*
+ * 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 com.google.common.base.Preconditions;
+import org.apache.druid.client.coordinator.NoopCoordinatorClient;
+import org.apache.druid.client.indexing.ClientCompactionTaskGranularitySpec;
+import org.apache.druid.data.input.InputFormat;
+import org.apache.druid.data.input.InputSource;
+import org.apache.druid.data.input.impl.CsvInputFormat;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.InlineInputSource;
+import org.apache.druid.data.input.impl.JsonInputFormat;
+import org.apache.druid.data.input.impl.LocalInputSource;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.indexer.granularity.GranularitySpec;
+import org.apache.druid.indexer.granularity.UniformGranularitySpec;
+import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.indexing.common.SegmentCacheManagerFactory;
+import org.apache.druid.indexing.common.config.TaskConfigBuilder;
+import org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexIOConfig;
+import org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexIngestionSpec;
+import org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexSupervisorTask;
+import org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexTuningConfig;
+import org.apache.druid.indexing.input.DruidInputSource;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.joda.time.Interval;
+
+import java.io.File;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Builder for the {@link Task} objects.
+ *
+ * The builder does not use any defaults and all required fields must be set
+ * explicitly.
+ *
+ * @param Type of task created by this builder.
+ * @param Type of tuning config used by this builder.
+ * @see #ofTypeIndex()
+ * @see #tuningConfig(Consumer) to specify the {@code tuningConfig}.
+ */
+@SuppressWarnings("unchecked")
+public abstract class TaskBuilder, C, T extends Task>
+{
+ // Fields are package-protected to allow access by subclasses like Index, Compact
+ InputSource inputSource = null;
+ InputFormat inputFormat = null;
+
+ final Map context = new HashMap<>();
+
+ Boolean appendToExisting = null;
+
+ final TuningConfigBuilder tuningConfig;
+
+ private TaskBuilder()
+ {
+ this.tuningConfig = tuningConfigBuilder();
+ }
+
+ /**
+ * Creates a raw Map-based payload for a {@code Task} that may be submitted to
+ * the Overlord using {@code OverlordClient.runTask()}.
+ */
+ public abstract T withId(String taskId);
+
+ public abstract B dataSource(String dataSource);
+
+ protected abstract TuningConfigBuilder tuningConfigBuilder();
+
+ /**
+ * Initializes builder for a new {@link IndexTask}.
+ */
+ public static Index ofTypeIndex()
+ {
+ return new Index();
+ }
+
+ /**
+ * Initializes builder for a new {@link ParallelIndexSupervisorTask}.
+ */
+ public static IndexParallel ofTypeIndexParallel()
+ {
+ return new IndexParallel();
+ }
+
+ public static Compact ofTypeCompact()
+ {
+ return new Compact();
+ }
+
+ public B inputSource(InputSource inputSource)
+ {
+ this.inputSource = inputSource;
+ return (B) this;
+ }
+
+ public B inlineInputSourceWithData(String data)
+ {
+ return inputSource(new InlineInputSource(data));
+ }
+
+ public B druidInputSource(String dataSource, Interval interval)
+ {
+ return inputSource(
+ new DruidInputSource(
+ dataSource,
+ interval,
+ null,
+ null,
+ null,
+ null,
+ TestHelper.getTestIndexIO(),
+ new NoopCoordinatorClient(),
+ new SegmentCacheManagerFactory(TestHelper.getTestIndexIO(), TestHelper.JSON_MAPPER),
+ new TaskConfigBuilder().build()
+ )
+ );
+ }
+
+ /**
+ * Gets the absolute path of the given resource files and sets:
+ *
+ * "inputSource": {
+ * "type": "local",
+ * "files": [<absolute-paths-of-given-resource-files>]
+ * }
+ *
+ */
+ public B localInputSourceWithFiles(String... resources)
+ {
+ try {
+ final List files = new ArrayList<>();
+ for (String file : resources) {
+ final URL resourceUrl = getClass().getClassLoader().getResource(file);
+ if (resourceUrl == null) {
+ throw new ISE("Could not find file[%s]", file);
+ }
+
+ files.add(new File(resourceUrl.toURI()));
+ }
+
+ return inputSource(
+ new LocalInputSource(null, null, files, null)
+ );
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public B inputFormat(InputFormat inputFormat)
+ {
+ this.inputFormat = inputFormat;
+ return (B) this;
+ }
+
+ public B jsonInputFormat()
+ {
+ return inputFormat(
+ new JsonInputFormat(null, null, null, null, null)
+ );
+ }
+
+ public B csvInputFormatWithColumns(String... columns)
+ {
+ return inputFormat(
+ new CsvInputFormat(List.of(columns), null, null, false, 0, null)
+ );
+ }
+
+ public B appendToExisting(boolean append)
+ {
+ this.appendToExisting = append;
+ return (B) this;
+ }
+
+ public B dynamicPartitionWithMaxRows(int maxRowsPerSegment)
+ {
+ tuningConfig.withPartitionsSpec(new DynamicPartitionsSpec(maxRowsPerSegment, null));
+ return (B) this;
+ }
+
+ public B tuningConfig(Consumer> updateTuningConfig)
+ {
+ updateTuningConfig.accept(tuningConfig);
+ return (B) this;
+ }
+
+ public B context(String key, Object value)
+ {
+ this.context.put(key, value);
+ return (B) this;
+ }
+
+ public abstract static class IndexCommon, C, T extends Task>
+ extends TaskBuilder
+ {
+ final DataSchema.Builder dataSchema = DataSchema.builder();
+
+ @Override
+ public B dataSource(String dataSource)
+ {
+ dataSchema.withDataSource(dataSource);
+ return (B) this;
+ }
+
+ public B dataSchema(Consumer updateDataSchema)
+ {
+ updateDataSchema.accept(dataSchema);
+ return (B) this;
+ }
+
+ public B isoTimestampColumn(String timestampColumn)
+ {
+ dataSchema.withTimestamp(new TimestampSpec(timestampColumn, "iso", null));
+ return (B) this;
+ }
+
+ public B timestampColumn(String timestampColumn)
+ {
+ dataSchema.withTimestamp(new TimestampSpec(timestampColumn, null, null));
+ return (B) this;
+ }
+
+ public B granularitySpec(GranularitySpec granularitySpec)
+ {
+ dataSchema.withGranularity(granularitySpec);
+ return (B) this;
+ }
+
+ public B granularitySpec(String segmentGranularity, String queryGranularity, Boolean rollup)
+ {
+ dataSchema.withGranularity(
+ new UniformGranularitySpec(
+ Granularity.fromString(segmentGranularity),
+ queryGranularity == null ? null : Granularity.fromString(queryGranularity),
+ rollup,
+ null
+ )
+ );
+ return (B) this;
+ }
+
+ /**
+ * Sets {@code "granularitySpec": {"segmentGranularity": }}.
+ */
+ public B segmentGranularity(String granularity)
+ {
+ return granularitySpec(granularity, null, null);
+ }
+
+ /**
+ * Sets the given dimensions as string dimensions in the {@link DataSchema}.
+ *
+ * @see #dataSchema(Consumer) for more options
+ */
+ public B dimensions(String... dimensions)
+ {
+ dataSchema.withDimensions(
+ Stream.of(dimensions)
+ .map(StringDimensionSchema::new)
+ .collect(Collectors.toList())
+ );
+ return (B) this;
+ }
+
+ public B metricAggregates(AggregatorFactory... aggregators)
+ {
+ dataSchema.withAggregators(aggregators);
+ return (B) this;
+ }
+ }
+
+ /**
+ * Builder for {@link IndexTask} that uses a {@link IndexTask.IndexTuningConfig}.
+ */
+ public static class Index extends IndexCommon
+ {
+ @Override
+ public TuningConfigBuilder tuningConfigBuilder()
+ {
+ return TuningConfigBuilder.forIndexTask();
+ }
+
+ @Override
+ public IndexTask withId(String taskId)
+ {
+ Preconditions.checkNotNull(inputSource, "'inputSource' must be specified");
+
+ return new IndexTask(
+ taskId,
+ null,
+ new IndexTask.IndexIngestionSpec(
+ dataSchema.build(),
+ new IndexTask.IndexIOConfig(
+ inputSource,
+ inputFormat,
+ appendToExisting,
+ null
+ ),
+ tuningConfig.build()
+ ),
+ context
+ );
+ }
+ }
+
+ /**
+ * Builder for {@link ParallelIndexSupervisorTask} which uses a {@link ParallelIndexTuningConfig}.
+ */
+ public static class IndexParallel extends IndexCommon
+ {
+ @Override
+ public ParallelIndexSupervisorTask withId(String taskId)
+ {
+ Preconditions.checkNotNull(inputSource, "'inputSource' must be specified");
+ return new ParallelIndexSupervisorTask(
+ taskId,
+ null,
+ null,
+ new ParallelIndexIngestionSpec(
+ dataSchema.build(),
+ new ParallelIndexIOConfig(
+ inputSource,
+ inputFormat,
+ appendToExisting,
+ null
+ ),
+ tuningConfig.build()
+ ),
+ context
+ );
+ }
+
+ @Override
+ public TuningConfigBuilder tuningConfigBuilder()
+ {
+ return TuningConfigBuilder.forParallelIndexTask();
+ }
+ }
+
+ /**
+ * Builder for a {@link CompactionTask} which uses a {@link CompactionTask.CompactionTuningConfig}.
+ */
+ public static class Compact extends TaskBuilder
+ {
+ private String dataSource;
+ private Interval interval;
+ private DimensionsSpec dimensionsSpec;
+ private Granularity segmentGranularity;
+ private ClientCompactionTaskGranularitySpec granularitySpec;
+ private CompactionIOConfig ioConfig;
+
+ @Override
+ public Compact dataSource(String dataSource)
+ {
+ this.dataSource = dataSource;
+ return this;
+ }
+
+ public Compact interval(Interval interval)
+ {
+ this.interval = interval;
+ return this;
+ }
+
+ public Compact dimensions(String... dimensions)
+ {
+ dimensionsSpec = new DimensionsSpec(
+ Stream.of(dimensions)
+ .map(StringDimensionSchema::new)
+ .collect(Collectors.toList())
+ );
+ return this;
+ }
+
+ public Compact granularitySpec(ClientCompactionTaskGranularitySpec granularitySpec)
+ {
+ this.granularitySpec = granularitySpec;
+ return this;
+ }
+
+ public Compact segmentGranularity(Granularity segmentGranularity)
+ {
+ this.segmentGranularity = segmentGranularity;
+ return this;
+ }
+
+ public Compact ioConfig(CompactionInputSpec inputSpec, boolean allowNonAlignedInterval)
+ {
+ this.ioConfig = new CompactionIOConfig(inputSpec, allowNonAlignedInterval, null);
+ return this;
+ }
+
+ @Override
+ public CompactionTask withId(String taskId)
+ {
+ return new CompactionTask(
+ taskId,
+ null,
+ dataSource,
+ interval,
+ null,
+ ioConfig,
+ dimensionsSpec,
+ null,
+ null,
+ null,
+ segmentGranularity,
+ granularitySpec,
+ null,
+ tuningConfig.build(),
+ null,
+ null,
+ null
+ );
+ }
+
+ @Override
+ public TuningConfigBuilder tuningConfigBuilder()
+ {
+ return TuningConfigBuilder.forCompactionTask();
+ }
+ }
+}
diff --git a/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/ITBestEffortRollupParallelIndexTest.java b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/ITBestEffortRollupParallelIndexTest.java
deleted file mode 100644
index 20bdd9767a77..000000000000
--- a/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/ITBestEffortRollupParallelIndexTest.java
+++ /dev/null
@@ -1,250 +0,0 @@
-/*
- * 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.testsEx.indexer;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.google.inject.Inject;
-import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
-import org.apache.druid.indexer.partitions.PartitionsSpec;
-import org.apache.druid.java.util.common.Pair;
-import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
-import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
-import org.apache.druid.testing.utils.ITRetryUtil;
-import org.apache.druid.testsEx.categories.BatchIndex;
-import org.apache.druid.testsEx.config.DruidTestRunner;
-import org.junit.Assert;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.runner.RunWith;
-
-import java.io.Closeable;
-import java.util.function.Function;
-
-@RunWith(DruidTestRunner.class)
-@Category(BatchIndex.class)
-public class ITBestEffortRollupParallelIndexTest extends AbstractITBatchIndexTest
-{
- // This ingestion spec has a splitHintSpec of maxSplitSize of 1 to test whether or not the task can handle
- // maxSplitSize of 1 properly.
- private static final String INDEX_TASK = "/indexer/wikipedia_parallel_index_task.json";
- private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_parallel_index_queries.json";
- private static final String REINDEX_TASK = "/indexer/wikipedia_parallel_reindex_task.json";
- private static final String REINDEX_QUERIES_RESOURCE = "/indexer/wikipedia_parallel_reindex_queries.json";
- private static final String INDEX_DATASOURCE = "wikipedia_parallel_index_test";
- private static final String INDEX_INGEST_SEGMENT_DATASOURCE = "wikipedia_parallel_ingest_segment_index_test";
- private static final String INDEX_INGEST_SEGMENT_TASK = "/indexer/wikipedia_parallel_ingest_segment_index_task.json";
- private static final String INDEX_DRUID_INPUT_SOURCE_DATASOURCE = "wikipedia_parallel_druid_input_source_index_test";
- private static final String INDEX_DRUID_INPUT_SOURCE_TASK = "/indexer/wikipedia_parallel_druid_input_source_index_task.json";
-
- private static final CoordinatorDynamicConfig DYNAMIC_CONFIG_PAUSED =
- CoordinatorDynamicConfig.builder().withPauseCoordination(true).build();
- private static final CoordinatorDynamicConfig DYNAMIC_CONFIG_DEFAULT =
- CoordinatorDynamicConfig.builder().build();
-
- @Inject
- CoordinatorResourceTestClient coordinatorClient;
-
- @Test
- public void testIndexData() throws Exception
- {
- PartitionsSpec partitionsSpec = new DynamicPartitionsSpec(null, null);
- try (
- final Closeable ignored1 = unloader(INDEX_DATASOURCE + config.getExtraDatasourceNameSuffix());
- final Closeable ignored2 = unloader(INDEX_INGEST_SEGMENT_DATASOURCE + config.getExtraDatasourceNameSuffix());
- final Closeable ignored3 = unloader(INDEX_DRUID_INPUT_SOURCE_DATASOURCE + config.getExtraDatasourceNameSuffix())
- ) {
- boolean forceGuaranteedRollup = partitionsSpec.isForceGuaranteedRollupCompatible();
- Assert.assertFalse("parititionSpec does not support best-effort rollup", forceGuaranteedRollup);
-
- final Function rollupTransform = spec -> {
- try {
- spec = StringUtils.replace(
- spec,
- "%%FORCE_GUARANTEED_ROLLUP%%",
- Boolean.toString(false)
- );
- spec = StringUtils.replace(
- spec,
- "%%SEGMENT_AVAIL_TIMEOUT_MILLIS%%",
- jsonMapper.writeValueAsString("0")
- );
- return StringUtils.replace(
- spec,
- "%%PARTITIONS_SPEC%%",
- jsonMapper.writeValueAsString(partitionsSpec)
- );
- }
- catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
- };
-
- doIndexTest(
- INDEX_DATASOURCE,
- INDEX_TASK,
- rollupTransform,
- INDEX_QUERIES_RESOURCE,
- false,
- true,
- true,
- new Pair<>(false, false)
- );
-
- // Index again, this time only choosing the second data file, and without explicit intervals chosen.
- // The second datafile covers both day segments, so this should replace them, as reflected in the queries.
- doIndexTest(
- INDEX_DATASOURCE,
- REINDEX_TASK,
- rollupTransform,
- REINDEX_QUERIES_RESOURCE,
- true,
- true,
- true,
- new Pair<>(false, false)
- );
-
- doReindexTest(
- INDEX_DATASOURCE,
- INDEX_INGEST_SEGMENT_DATASOURCE,
- rollupTransform,
- INDEX_INGEST_SEGMENT_TASK,
- REINDEX_QUERIES_RESOURCE,
- new Pair<>(false, false)
- );
-
- // with DruidInputSource
- doReindexTest(
- INDEX_DATASOURCE,
- INDEX_DRUID_INPUT_SOURCE_DATASOURCE,
- rollupTransform,
- INDEX_DRUID_INPUT_SOURCE_TASK,
- REINDEX_QUERIES_RESOURCE,
- new Pair<>(false, false)
- );
- }
- }
-
- /**
- * Test a non zero value for awaitSegmentAvailabilityTimeoutMillis. This will confirm that the report for the task
- * indicates segments were confirmed to be available on the cluster before finishing the ingestion job.
- */
- @Test
- public void testIndexDataVerifySegmentAvailability() throws Exception
- {
- PartitionsSpec partitionsSpec = new DynamicPartitionsSpec(null, null);
- try (
- final Closeable ignored1 = unloader(INDEX_DATASOURCE + config.getExtraDatasourceNameSuffix());
- ) {
- boolean forceGuaranteedRollup = partitionsSpec.isForceGuaranteedRollupCompatible();
- Assert.assertFalse("parititionSpec does not support best-effort rollup", forceGuaranteedRollup);
-
- final Function rollupTransform = spec -> {
- try {
- spec = StringUtils.replace(
- spec,
- "%%FORCE_GUARANTEED_ROLLUP%%",
- Boolean.toString(false)
- );
- spec = StringUtils.replace(
- spec,
- "%%SEGMENT_AVAIL_TIMEOUT_MILLIS%%",
- jsonMapper.writeValueAsString("600000")
- );
- return StringUtils.replace(
- spec,
- "%%PARTITIONS_SPEC%%",
- jsonMapper.writeValueAsString(partitionsSpec)
- );
- }
- catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
- };
-
- doIndexTest(
- INDEX_DATASOURCE,
- INDEX_TASK,
- rollupTransform,
- INDEX_QUERIES_RESOURCE,
- false,
- true,
- true,
- new Pair<>(true, true)
- );
- }
- }
-
- /**
- * Test a non zero value for awaitSegmentAvailabilityTimeoutMillis. Setting the config value to 1 millis
- * and pausing coordination to confirm that the task will still succeed even if the job was not able to confirm the
- * segments were loaded by the time the timeout occurs.
- */
- @Test
- public void testIndexDataAwaitSegmentAvailabilityFailsButTaskSucceeds() throws Exception
- {
- PartitionsSpec partitionsSpec = new DynamicPartitionsSpec(null, null);
- try (
- final Closeable ignored1 = unloader(INDEX_DATASOURCE + config.getExtraDatasourceNameSuffix());
- ) {
- coordinatorClient.postDynamicConfig(DYNAMIC_CONFIG_PAUSED);
- boolean forceGuaranteedRollup = partitionsSpec.isForceGuaranteedRollupCompatible();
- Assert.assertFalse("parititionSpec does not support best-effort rollup", forceGuaranteedRollup);
-
- final Function rollupTransform = spec -> {
- try {
- spec = StringUtils.replace(
- spec,
- "%%FORCE_GUARANTEED_ROLLUP%%",
- Boolean.toString(false)
- );
- spec = StringUtils.replace(
- spec,
- "%%SEGMENT_AVAIL_TIMEOUT_MILLIS%%",
- jsonMapper.writeValueAsString("1")
- );
- return StringUtils.replace(
- spec,
- "%%PARTITIONS_SPEC%%",
- jsonMapper.writeValueAsString(partitionsSpec)
- );
- }
- catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
- };
-
- doIndexTest(
- INDEX_DATASOURCE,
- INDEX_TASK,
- rollupTransform,
- INDEX_QUERIES_RESOURCE,
- false,
- false,
- false,
- new Pair<>(true, false)
- );
- coordinatorClient.postDynamicConfig(DYNAMIC_CONFIG_DEFAULT);
- ITRetryUtil.retryUntilTrue(
- () -> coordinator.areSegmentsLoaded(INDEX_DATASOURCE + config.getExtraDatasourceNameSuffix()), "Segment Load"
- );
- }
- }
-}
diff --git a/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_index_queries.json b/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_index_queries.json
deleted file mode 100644
index 193e69fbc9ca..000000000000
--- a/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_index_queries.json
+++ /dev/null
@@ -1,80 +0,0 @@
-[
- {
- "description": "timeseries, 1 agg, all",
- "query":{
- "queryType" : "timeBoundary",
- "dataSource": "%%DATASOURCE%%"
- },
- "expectedResults":[
- {
- "timestamp" : "2015-09-12T00:00:00.000Z",
- "result" : {
- "minTime" : "2015-09-12T00:00:00.000Z",
- "maxTime" : "2015-09-12T00:00:00.000Z"
- }
- }
- ]
- },
- {
- "description": "scan, all",
- "query": {
- "queryType": "scan",
- "dataSource": "%%DATASOURCE%%",
- "intervals": [
- "2013-01-01/2020-01-02"
- ],
- "resultFormat":"compactedList"
- },
- "expectedResults": %%EXPECTED_SCAN_RESULT%%,
- "fieldsToTest": ["events"]
- },
- {
- "description": "roll up ratio",
- "query": {
- "queryType":"timeseries",
- "dataSource":{
- "type":"table",
- "name":"%%DATASOURCE%%"
- },
- "intervals":{
- "type":"intervals",
- "intervals":[
- "2013-01-01/2020-01-02"
- ]
- },
- "granularity":{
- "type":"all"
- },
- "aggregations":[
- {
- "type":"count",
- "name":"a0"
- },
- {
- "type":"longSum",
- "name":"a1",
- "fieldName":"count",
- "expression":null
- }
- ],
- "postAggregations":[
- {
- "type":"expression",
- "name":"p0",
- "expression":"((\"a0\" * 1.00) / \"a1\")",
- "ordering":null
- }
- ]
- },
- "expectedResults": [
- {
- "timestamp" : "2015-09-12T00:00:00.000Z",
- "result" : {
- "a1" : %%EXPECTED_SUM_COUNT%%,
- "p0" : %%EXPECTED_ROLLUP_RATIO%%,
- "a0" : %%EXPECTED_NUM_ROW%%
- }
- }
- ]
- }
-]
\ No newline at end of file
diff --git a/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_index_task.json b/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_index_task.json
deleted file mode 100644
index 3a21a856ac6f..000000000000
--- a/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_index_task.json
+++ /dev/null
@@ -1,57 +0,0 @@
-{
- "type": "index_parallel",
- "spec": {
- "ioConfig": {
- "type": "index_parallel",
- "inputSource": {
- "type": "inline",
- "data": "{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"C\",\"dimB\":\"F\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"C\",\"dimB\":\"J\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"H\",\"dimB\":\"X\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"Z\",\"dimB\":\"S\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"H\",\"dimB\":\"X\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"H\",\"dimB\":\"Z\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"J\",\"dimB\":\"R\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"H\",\"dimB\":\"T\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"H\",\"dimB\":\"X\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimC\":\"A\",\"dimB\":\"X\",\"metA\":1}\n"
- },
- "inputFormat": {
- "type": "json"
- }
- },
- "tuningConfig": {
- "type": "index_parallel",
- "partitionsSpec": {
- "type": "dynamic",
- "maxRowsPerSegment": 3,
- "maxTotalRows": 3
- },
- "maxRowsInMemory": 3
- },
- "dataSchema": {
- "dataSource": "%%DATASOURCE%%",
- "timestampSpec": {
- "column": "time",
- "format": "iso"
- },
- "dimensionsSpec": {
- "dimensions": [
- "dimB",
- "dimA",
- "dimC",
- "dimD",
- "dimE",
- "dimF"
- ]
- },
- "granularitySpec": {
- "queryGranularity": "hour",
- "rollup": true,
- "segmentGranularity": "hour"
- },
- "metricsSpec": [
- {
- "name": "count",
- "type": "count"
- },
- {
- "name": "sum_metA",
- "type": "longSum",
- "fieldName": "metA"
- }
- ]
- }
- }
-}
\ No newline at end of file
diff --git a/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_with_dim_compaction_task.json b/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_with_dim_compaction_task.json
deleted file mode 100644
index 9416a3f6bdad..000000000000
--- a/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_with_dim_compaction_task.json
+++ /dev/null
@@ -1,19 +0,0 @@
-{
- "type": "compact",
- "dataSource": "%%DATASOURCE%%",
- "dimensionsSpec": {
- "dimensions": %%DIMENSION_NAMES%%
- },
- "interval": "2010-10-29T05:00:00Z/2030-10-29T06:00:00Z",
- "tuningConfig": {
- "type": "index_parallel",
- "maxRowsPerSegment": 3,
- "maxRowsInMemory": 3,
- "maxNumConcurrentSubTasks": 2,
- "partitionsSpec": {
- "type": "hashed",
- "numShards": 1
- },
- "forceGuaranteedRollup": true
- }
-}
\ No newline at end of file
diff --git a/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_without_dim_compaction_task.json b/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_without_dim_compaction_task.json
deleted file mode 100644
index a149d7a25127..000000000000
--- a/integration-tests-ex/cases/src/test/resources/indexer/sparse_column_without_dim_compaction_task.json
+++ /dev/null
@@ -1,16 +0,0 @@
-{
- "type": "compact",
- "dataSource": "%%DATASOURCE%%",
- "interval": "2010-10-29T05:00:00Z/2030-10-29T06:00:00Z",
- "tuningConfig": {
- "type": "index_parallel",
- "maxRowsPerSegment": 3,
- "maxRowsInMemory": 3,
- "maxNumConcurrentSubTasks": 2,
- "partitionsSpec": {
- "type": "hashed",
- "numShards": 1
- },
- "forceGuaranteedRollup": true
- }
-}
\ No newline at end of file
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/TestNGGroup.java b/integration-tests/src/test/java/org/apache/druid/tests/TestNGGroup.java
index 9ba40a8d1d89..601754a27007 100644
--- a/integration-tests/src/test/java/org/apache/druid/tests/TestNGGroup.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/TestNGGroup.java
@@ -41,14 +41,8 @@ public class TestNGGroup
public static final String KAFKA_DATA_FORMAT = "kafka-data-format";
- public static final String COMPACTION = "compaction";
-
- public static final String UPGRADE = "upgrade";
-
public static final String APPEND_INGESTION = "append-ingestion";
- public static final String PERFECT_ROLLUP_PARALLEL_BATCH_INDEX = "perfect-rollup-parallel-batch-index";
-
/**
* This group can only be run individually using -Dgroups=query since it requires specific test data setup.
*/
@@ -152,12 +146,8 @@ public class TestNGGroup
public static final String HIGH_AVAILABILTY = "high-availability";
- public static final String SHUFFLE_DEEP_STORE = "shuffle-deep-store";
-
public static final String CUSTOM_COORDINATOR_DUTIES = "custom-coordinator-duties";
- public static final String HTTP_ENDPOINT = "http-endpoint";
-
public static final String CENTRALIZED_DATASOURCE_SCHEMA = "centralized-datasource-schema";
public static final String CDS_TASK_SCHEMA_PUBLISH_DISABLED = "cds-task-schema-publish-disabled";
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/api/ITOverlordResourceNotFoundTest.java b/integration-tests/src/test/java/org/apache/druid/tests/api/ITOverlordResourceNotFoundTest.java
deleted file mode 100644
index 8959c7744fe0..000000000000
--- a/integration-tests/src/test/java/org/apache/druid/tests/api/ITOverlordResourceNotFoundTest.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/*
- * 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.tests.api;
-
-import com.google.inject.Inject;
-import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.testing.clients.OverlordResourceTestClient;
-import org.apache.druid.testing.guice.DruidTestModuleFactory;
-import org.apache.druid.tests.TestNGGroup;
-import org.testng.Assert;
-import org.testng.annotations.Guice;
-import org.testng.annotations.Test;
-
-import java.util.function.Consumer;
-
-@Test(groups = TestNGGroup.HTTP_ENDPOINT)
-@Guice(moduleFactory = DruidTestModuleFactory.class)
-public class ITOverlordResourceNotFoundTest
-{
- @Inject
- protected OverlordResourceTestClient indexer;
-
- @Test
- public void testGetSupervisorStatusNotFound()
- {
- callAndCheckNotFound(indexer::getSupervisorStatus);
- }
-
- @Test
- public void testGetSupervisorHistoryNotFound()
- {
- callAndCheckNotFound(indexer::getSupervisorHistory);
- }
-
- @Test
- public void testResumeSupervisorNotFound()
- {
- callAndCheckNotFound(indexer::resumeSupervisor);
- }
-
- @Test
- public void testSuspendSupervisorNotFound()
- {
- callAndCheckNotFound(indexer::suspendSupervisor);
- }
-
- @Test
- public void testShutdownSupervisorNotFound()
- {
- callAndCheckNotFound(indexer::shutdownSupervisor);
- }
-
- @Test
- public void testTerminateSupervisorNotFound()
- {
- callAndCheckNotFound(indexer::terminateSupervisor);
- }
-
- @Test
- public void testGetSupervisorHealthNotFound()
- {
- callAndCheckNotFound(indexer::getSupervisorHealth);
- }
-
- @Test
- public void testStatsSupervisorNotFound()
- {
- callAndCheckNotFound(indexer::statsSupervisor);
- }
-
- @Test
- public void testResetSupervisorNotFound()
- {
- callAndCheckNotFound(indexer::resetSupervisor);
- }
-
- @Test
- public void testGetTaskStatusNotFound()
- {
- callAndCheckNotFound(indexer::getTaskStatus);
- }
-
- @Test
- public void testShutdownTaskNotFound()
- {
- callAndCheckNotFound(indexer::shutdownTask);
- }
-
- @Test
- public void testGetTaskLogNotFound()
- {
- callAndCheckNotFound(indexer::getTaskLog);
- }
-
- @Test
- public void testGetTaskReportNotFound()
- {
- callAndCheckNotFound(indexer::getTaskReport);
- }
-
- @Test
- public void testGetTaskPayLoadNotFound()
- {
- callAndCheckNotFound(indexer::getTaskPayload);
- }
-
- private void callAndCheckNotFound(Consumer runnable)
- {
- String supervisorId = "not_exist_id";
- try {
- runnable.accept(supervisorId);
- }
- catch (ISE e) {
- // OverlordResourceTestClient turns all non-200 response into ISE exception
- // So we catch ISE and check if the message in this exception matches expected message
- Assert.assertTrue(
- e.getMessage().contains("[404 Not Found") && e.getMessage().contains(supervisorId),
- "Unexpected exception. Message does not match expected. " + e.getMessage()
- );
- return;
- }
- Assert.fail("Should not go to here");
- }
-}
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/api/ITOverlordResourceTest.java b/integration-tests/src/test/java/org/apache/druid/tests/api/ITOverlordResourceTest.java
deleted file mode 100644
index eb59afee0f76..000000000000
--- a/integration-tests/src/test/java/org/apache/druid/tests/api/ITOverlordResourceTest.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * 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.tests.api;
-
-import com.google.inject.Inject;
-import org.apache.druid.indexer.TaskState;
-import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.testing.clients.OverlordResourceTestClient;
-import org.apache.druid.testing.clients.TaskResponseObject;
-import org.apache.druid.testing.guice.DruidTestModuleFactory;
-import org.apache.druid.testing.utils.ITRetryUtil;
-import org.apache.druid.tests.TestNGGroup;
-import org.apache.druid.tests.indexer.AbstractIndexerTest;
-import org.testng.annotations.Guice;
-import org.testng.annotations.Test;
-
-import java.io.IOException;
-import java.util.List;
-
-@Test(groups = TestNGGroup.HTTP_ENDPOINT)
-@Guice(moduleFactory = DruidTestModuleFactory.class)
-public class ITOverlordResourceTest
-{
- private static final String INGESTION_SPEC = "/api/overlord-resource-test-task.json";
-
- @Inject
- protected OverlordResourceTestClient indexer;
-
- @Test
- public void testGetAllTasks() throws IOException
- {
- final String taskSpec = AbstractIndexerTest.getResourceAsString(INGESTION_SPEC);
- final String taskId = indexer.submitTask(taskSpec);
-
- ITRetryUtil.retryUntil(
- () -> {
- final List tasks = indexer.getAllTasks();
- final TaskResponseObject taskStatus = tasks
- .stream()
- .filter(task -> taskId.equals(task.getId()))
- .findAny()
- .orElseThrow(() -> new ISE("Cannot find task[%s]", taskId));
- TaskState status = taskStatus.getStatus();
- if (status == TaskState.FAILED) {
- throw new ISE("Task[%s] FAILED", taskId);
- }
- return status == TaskState.SUCCESS;
- },
- true,
- ITRetryUtil.DEFAULT_RETRY_SLEEP,
- ITRetryUtil.DEFAULT_RETRY_COUNT,
- taskId
- );
- }
-}
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionLockContentionTest.java b/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionLockContentionTest.java
deleted file mode 100644
index 11e4ec6cad30..000000000000
--- a/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionLockContentionTest.java
+++ /dev/null
@@ -1,361 +0,0 @@
-/*
- * 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.tests.coordinator.duty;
-
-import com.google.inject.Inject;
-import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
-import org.apache.druid.java.util.common.Intervals;
-import org.apache.druid.java.util.common.logger.Logger;
-import org.apache.druid.metadata.LockFilterPolicy;
-import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
-import org.apache.druid.server.coordinator.ClusterCompactionConfig;
-import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
-import org.apache.druid.server.coordinator.InlineSchemaDataSourceCompactionConfig;
-import org.apache.druid.testing.clients.CompactionResourceTestClient;
-import org.apache.druid.testing.clients.TaskResponseObject;
-import org.apache.druid.testing.guice.DruidTestModuleFactory;
-import org.apache.druid.testing.utils.EventSerializer;
-import org.apache.druid.testing.utils.ITRetryUtil;
-import org.apache.druid.testing.utils.KafkaUtil;
-import org.apache.druid.testing.utils.StreamEventWriter;
-import org.apache.druid.testing.utils.StreamGenerator;
-import org.apache.druid.testing.utils.WikipediaStreamEventStreamGenerator;
-import org.apache.druid.tests.TestNGGroup;
-import org.apache.druid.tests.indexer.AbstractKafkaIndexingServiceTest;
-import org.apache.druid.tests.indexer.AbstractStreamIndexingTest;
-import org.apache.druid.timeline.DataSegment;
-import org.joda.time.Interval;
-import org.joda.time.Period;
-import org.testng.Assert;
-import org.testng.annotations.BeforeClass;
-import org.testng.annotations.BeforeMethod;
-import org.testng.annotations.DataProvider;
-import org.testng.annotations.Guice;
-import org.testng.annotations.Test;
-
-import java.io.Closeable;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-/**
- * Integration Test to verify behaviour when there is a lock contention between
- * compaction tasks and on-going stream ingestion tasks.
- */
-@Test(groups = {TestNGGroup.COMPACTION})
-@Guice(moduleFactory = DruidTestModuleFactory.class)
-public class ITAutoCompactionLockContentionTest extends AbstractKafkaIndexingServiceTest
-{
- private static final Logger LOG = new Logger(ITAutoCompactionLockContentionTest.class);
-
- @Inject
- private CompactionResourceTestClient compactionResource;
-
- private GeneratedTestConfig generatedTestConfig;
- private StreamGenerator streamGenerator;
-
- private String fullDatasourceName;
-
- @DataProvider
- public static Object[] getParameters()
- {
- return new Object[]{false, true};
- }
-
- @BeforeClass
- public void setupClass() throws Exception
- {
- doBeforeClass();
- }
-
- @BeforeMethod
- public void setup() throws Exception
- {
- generatedTestConfig = new GeneratedTestConfig(
- Specs.PARSER_TYPE,
- getResourceAsString(Specs.INPUT_FORMAT_PATH)
- );
- fullDatasourceName = generatedTestConfig.getFullDatasourceName();
- final EventSerializer serializer = jsonMapper.readValue(
- getResourceAsStream(Specs.SERIALIZER_PATH),
- EventSerializer.class
- );
- streamGenerator = new WikipediaStreamEventStreamGenerator(serializer, 6, 100);
- }
-
- @Override
- public String getTestNamePrefix()
- {
- return "autocompact_lock_contention";
- }
-
- @Test(dataProvider = "getParameters")
- public void testAutoCompactionSkipsLockedIntervals(boolean transactionEnabled) throws Exception
- {
- if (shouldSkipTest(transactionEnabled)) {
- return;
- }
-
- try (
- final Closeable closer = createResourceCloser(generatedTestConfig);
- final StreamEventWriter streamEventWriter = createStreamEventWriter(config, transactionEnabled)
- ) {
- // Start supervisor
- final String taskSpec = generatedTestConfig.getStreamIngestionPropsTransform()
- .apply(getResourceAsString(SUPERVISOR_SPEC_TEMPLATE_PATH));
- generatedTestConfig.setSupervisorId(indexer.submitSupervisor(taskSpec));
- LOG.info("supervisorSpec: [%s]", taskSpec);
-
- // Generate data for minutes 1, 2 and 3
- final Interval minute1 = Intervals.of("2000-01-01T01:01:00Z/2000-01-01T01:02:00Z");
- final long rowsForMinute1 = generateData(minute1, streamEventWriter);
-
- final Interval minute2 = Intervals.of("2000-01-01T01:02:00Z/2000-01-01T01:03:00Z");
- long rowsForMinute2 = generateData(minute2, streamEventWriter);
-
- final Interval minute3 = Intervals.of("2000-01-01T01:03:00Z/2000-01-01T01:04:00Z");
- final long rowsForMinute3 = generateData(minute3, streamEventWriter);
-
- // Wait for data to be ingested for all the minutes
- ensureRowCount(rowsForMinute1 + rowsForMinute2 + rowsForMinute3);
-
- // Wait for the segments to be loaded and interval locks to be released
- ensureLockedIntervals();
- ensureSegmentsLoaded();
-
- // 2 segments for each minute, total 6
- ensureSegmentsCount(6);
-
- // Generate more data for minute2 so that it gets locked
- rowsForMinute2 += generateData(minute2, streamEventWriter);
- ensureLockedIntervals(minute2);
-
- // Trigger auto compaction
- submitAndVerifyCompactionConfig();
- compactionResource.forceTriggerAutoCompaction();
-
- // Wait for segments to be loaded
- ensureRowCount(rowsForMinute1 + rowsForMinute2 + rowsForMinute3);
- ensureLockedIntervals();
- ensureSegmentsLoaded();
-
- // Verify that minute1 and minute3 have been compacted
- ensureCompactionTaskCount(2);
- verifyCompactedIntervals(minute1, minute3);
-
- // Trigger auto compaction again
- compactionResource.forceTriggerAutoCompaction();
-
- // Verify that all the segments are now compacted
- ensureCompactionTaskCount(3);
- ensureSegmentsLoaded();
- verifyCompactedIntervals(minute1, minute2, minute3);
- ensureSegmentsCount(3);
- }
- }
-
- /**
- * Retries until the segment count is as expected.
- */
- private void ensureSegmentsCount(int numExpectedSegments)
- {
- ITRetryUtil.retryUntilEquals(
- () -> coordinator.getFullSegmentsMetadata(fullDatasourceName).size(),
- numExpectedSegments,
- "Segment count"
- );
- }
-
- /**
- * Verifies that the given intervals have been compacted.
- */
- private void verifyCompactedIntervals(Interval... compactedIntervals)
- {
- List segments = coordinator.getFullSegmentsMetadata(fullDatasourceName);
- List observedCompactedSegments = new ArrayList<>();
- Set observedCompactedIntervals = new HashSet<>();
- for (DataSegment segment : segments) {
- if (segment.getLastCompactionState() != null) {
- observedCompactedSegments.add(segment);
- observedCompactedIntervals.add(segment.getInterval());
- }
- }
-
- Set expectedCompactedIntervals = new HashSet<>(Arrays.asList(compactedIntervals));
- Assert.assertEquals(observedCompactedIntervals, expectedCompactedIntervals);
-
- DynamicPartitionsSpec expectedPartitionSpec = new DynamicPartitionsSpec(
- Specs.MAX_ROWS_PER_SEGMENT,
- Long.MAX_VALUE
- );
- for (DataSegment compactedSegment : observedCompactedSegments) {
- Assert.assertNotNull(compactedSegment.getLastCompactionState());
- Assert.assertEquals(
- compactedSegment.getLastCompactionState().getPartitionsSpec(),
- expectedPartitionSpec
- );
- }
- }
-
- /**
- * Generates data points for the specified interval.
- *
- * @return Number of rows generated.
- */
- private long generateData(Interval interval, StreamEventWriter streamEventWriter)
- {
- long rowCount = streamGenerator.run(
- generatedTestConfig.getStreamName(),
- streamEventWriter,
- 10,
- interval.getStart()
- );
- LOG.info("Generated %d Rows for Interval [%s]", rowCount, interval);
-
- return rowCount;
- }
-
- /**
- * Retries until segments have been loaded.
- */
- private void ensureSegmentsLoaded()
- {
- ITRetryUtil.retryUntilTrue(
- () -> coordinator.areSegmentsLoaded(fullDatasourceName),
- "Segments are loaded"
- );
- }
-
- /**
- * Retries until the specified Intervals are locked for the current datasource.
- * If no interval has been specified, retries until no interval is locked
- */
- private void ensureLockedIntervals(Interval... intervals)
- {
- final LockFilterPolicy lockFilterPolicy = new LockFilterPolicy(fullDatasourceName, 0, null, null);
- final Set expectedLockedIntervals = Arrays.stream(intervals).collect(Collectors.toSet());
- ITRetryUtil.retryUntilEquals(
- () -> Set.copyOf(
- indexer.getLockedIntervals(List.of(lockFilterPolicy))
- .getOrDefault(fullDatasourceName, List.of())
- ),
- expectedLockedIntervals,
- "Locked intervals"
- );
- }
-
- /**
- * Checks if a test should be skipped based on whether transaction is enabled or not.
- */
- private boolean shouldSkipTest(boolean testEnableTransaction)
- {
- Map kafkaTestProps = KafkaUtil
- .getAdditionalKafkaTestConfigFromProperties(config);
- boolean configEnableTransaction = Boolean.parseBoolean(
- kafkaTestProps.getOrDefault(KafkaUtil.TEST_CONFIG_TRANSACTION_ENABLED, "false")
- );
-
- return configEnableTransaction != testEnableTransaction;
- }
-
- /**
- * Submits a compaction config for the current datasource.
- */
- private void submitAndVerifyCompactionConfig() throws Exception
- {
- final DataSourceCompactionConfig dataSourceCompactionConfig = InlineSchemaDataSourceCompactionConfig
- .builder()
- .forDataSource(fullDatasourceName)
- .withSkipOffsetFromLatest(Period.ZERO)
- .withMaxRowsPerSegment(Specs.MAX_ROWS_PER_SEGMENT)
- .build();
- compactionResource.updateClusterConfig(new ClusterCompactionConfig(0.5, 10, null, null, null));
- compactionResource.submitCompactionConfig(dataSourceCompactionConfig);
-
- // Verify that the compaction config is updated correctly.
- DataSourceCompactionConfig observedCompactionConfig
- = compactionResource.getDataSourceCompactionConfig(fullDatasourceName);
- Assert.assertEquals(observedCompactionConfig, dataSourceCompactionConfig);
- }
-
- /**
- * Checks if the given TaskResponseObject represents a Compaction Task.
- */
- private boolean isCompactionTask(TaskResponseObject taskResponse)
- {
- return "compact".equalsIgnoreCase(taskResponse.getType());
- }
-
- /**
- * Retries until the total number of complete compaction tasks is as expected.
- */
- private void ensureCompactionTaskCount(int expectedCount)
- {
- ITRetryUtil.retryUntilEquals(
- this::getNumberOfCompletedCompactionTasks,
- expectedCount,
- "Number of completed compaction tasks"
- );
- }
-
- /**
- * Gets the number of complete compaction tasks.
- */
- private int getNumberOfCompletedCompactionTasks()
- {
- List completeTasks = indexer
- .getCompleteTasksForDataSource(fullDatasourceName);
-
- return (int) completeTasks.stream().filter(this::isCompactionTask).count();
- }
-
- /**
- * Retries until the total row count is as expected.
- */
- private void ensureRowCount(long totalRows)
- {
- ITRetryUtil.retryUntilEquals(
- () -> queryHelper.countRows(
- fullDatasourceName,
- Intervals.ETERNITY,
- name -> new LongSumAggregatorFactory(name, "count")
- ),
- totalRows,
- "Total row count in datasource"
- );
- }
-
- /**
- * Constants for test specs.
- */
- private static class Specs
- {
- static final String SERIALIZER_PATH = DATA_RESOURCE_ROOT + "/csv/serializer/serializer.json";
- static final String INPUT_FORMAT_PATH = DATA_RESOURCE_ROOT + "/csv/input_format/input_format.json";
- static final String PARSER_TYPE = AbstractStreamIndexingTest.INPUT_FORMAT;
-
- static final int MAX_ROWS_PER_SEGMENT = 10000;
- }
-
-}
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractIndexerTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractIndexerTest.java
index c35490336840..a4f8b10bad54 100644
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractIndexerTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractIndexerTest.java
@@ -211,7 +211,7 @@ public static String getResourceAsString(String file) throws IOException
public static InputStream getResourceAsStream(String resource)
{
- return ITCompactionTaskTest.class.getResourceAsStream(resource);
+ return AbstractIndexerTest.class.getResourceAsStream(resource);
}
public static List listResources(String dir) throws IOException
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITBestEffortRollupParallelIndexTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITBestEffortRollupParallelIndexTest.java
deleted file mode 100644
index 53deb831a2db..000000000000
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITBestEffortRollupParallelIndexTest.java
+++ /dev/null
@@ -1,261 +0,0 @@
-/*
- * 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.tests.indexer;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.google.inject.Inject;
-import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
-import org.apache.druid.indexer.partitions.PartitionsSpec;
-import org.apache.druid.java.util.common.Pair;
-import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
-import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
-import org.apache.druid.testing.guice.DruidTestModuleFactory;
-import org.apache.druid.testing.utils.ITRetryUtil;
-import org.apache.druid.tests.TestNGGroup;
-import org.testng.Assert;
-import org.testng.annotations.DataProvider;
-import org.testng.annotations.Guice;
-import org.testng.annotations.Test;
-
-import java.io.Closeable;
-import java.util.function.Function;
-
-@Test(groups = {TestNGGroup.BATCH_INDEX})
-@Guice(moduleFactory = DruidTestModuleFactory.class)
-public class ITBestEffortRollupParallelIndexTest extends AbstractITBatchIndexTest
-{
- // This ingestion spec has a splitHintSpec of maxSplitSize of 1 to test whether or not the task can handle
- // maxSplitSize of 1 properly.
- private static final String INDEX_TASK = "/indexer/wikipedia_parallel_index_task.json";
- private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_parallel_index_queries.json";
- private static final String REINDEX_TASK = "/indexer/wikipedia_parallel_reindex_task.json";
- private static final String REINDEX_QUERIES_RESOURCE = "/indexer/wikipedia_parallel_reindex_queries.json";
- private static final String INDEX_DATASOURCE = "wikipedia_parallel_index_test";
- private static final String INDEX_INGEST_SEGMENT_DATASOURCE = "wikipedia_parallel_ingest_segment_index_test";
- private static final String INDEX_INGEST_SEGMENT_TASK = "/indexer/wikipedia_parallel_ingest_segment_index_task.json";
- private static final String INDEX_DRUID_INPUT_SOURCE_DATASOURCE = "wikipedia_parallel_druid_input_source_index_test";
- private static final String INDEX_DRUID_INPUT_SOURCE_TASK = "/indexer/wikipedia_parallel_druid_input_source_index_task.json";
-
- private static final CoordinatorDynamicConfig DYNAMIC_CONFIG_PAUSED =
- CoordinatorDynamicConfig.builder().withPauseCoordination(true).build();
- private static final CoordinatorDynamicConfig DYNAMIC_CONFIG_DEFAULT =
- CoordinatorDynamicConfig.builder().build();
-
- @Inject
- CoordinatorResourceTestClient coordinatorClient;
-
- @DataProvider
- public static Object[][] resources()
- {
- return new Object[][]{
- {new DynamicPartitionsSpec(null, null)}
- };
- }
-
- @Test(dataProvider = "resources")
- public void testIndexData(PartitionsSpec partitionsSpec) throws Exception
- {
- try (
- final Closeable ignored1 = unloader(INDEX_DATASOURCE + config.getExtraDatasourceNameSuffix());
- final Closeable ignored2 = unloader(INDEX_INGEST_SEGMENT_DATASOURCE + config.getExtraDatasourceNameSuffix());
- final Closeable ignored3 = unloader(INDEX_DRUID_INPUT_SOURCE_DATASOURCE + config.getExtraDatasourceNameSuffix())
- ) {
- boolean forceGuaranteedRollup = partitionsSpec.isForceGuaranteedRollupCompatible();
- Assert.assertFalse(forceGuaranteedRollup, "parititionSpec does not support best-effort rollup");
-
- final Function rollupTransform = spec -> {
- try {
- spec = StringUtils.replace(
- spec,
- "%%FORCE_GUARANTEED_ROLLUP%%",
- Boolean.toString(false)
- );
- spec = StringUtils.replace(
- spec,
- "%%SEGMENT_AVAIL_TIMEOUT_MILLIS%%",
- jsonMapper.writeValueAsString("0")
- );
- return StringUtils.replace(
- spec,
- "%%PARTITIONS_SPEC%%",
- jsonMapper.writeValueAsString(partitionsSpec)
- );
- }
- catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
- };
-
- doIndexTest(
- INDEX_DATASOURCE,
- INDEX_TASK,
- rollupTransform,
- INDEX_QUERIES_RESOURCE,
- false,
- true,
- true,
- new Pair<>(false, false)
- );
-
- // Index again, this time only choosing the second data file, and without explicit intervals chosen.
- // The second datafile covers both day segments, so this should replace them, as reflected in the queries.
- doIndexTest(
- INDEX_DATASOURCE,
- REINDEX_TASK,
- rollupTransform,
- REINDEX_QUERIES_RESOURCE,
- true,
- true,
- true,
- new Pair<>(false, false)
- );
-
- doReindexTest(
- INDEX_DATASOURCE,
- INDEX_INGEST_SEGMENT_DATASOURCE,
- rollupTransform,
- INDEX_INGEST_SEGMENT_TASK,
- REINDEX_QUERIES_RESOURCE,
- new Pair<>(false, false)
- );
-
- // with DruidInputSource
- doReindexTest(
- INDEX_DATASOURCE,
- INDEX_DRUID_INPUT_SOURCE_DATASOURCE,
- rollupTransform,
- INDEX_DRUID_INPUT_SOURCE_TASK,
- REINDEX_QUERIES_RESOURCE,
- new Pair<>(false, false)
- );
- }
- }
-
- /**
- * Test a non zero value for awaitSegmentAvailabilityTimeoutMillis. This will confirm that the report for the task
- * indicates segments were confirmed to be available on the cluster before finishing the ingestion job.
- *
- * @param partitionsSpec
- * @throws Exception
- */
- @Test(dataProvider = "resources")
- public void testIndexDataVerifySegmentAvailability(PartitionsSpec partitionsSpec) throws Exception
- {
- try (
- final Closeable ignored1 = unloader(INDEX_DATASOURCE + config.getExtraDatasourceNameSuffix());
- ) {
- boolean forceGuaranteedRollup = partitionsSpec.isForceGuaranteedRollupCompatible();
- Assert.assertFalse(forceGuaranteedRollup, "parititionSpec does not support best-effort rollup");
-
- final Function rollupTransform = spec -> {
- try {
- spec = StringUtils.replace(
- spec,
- "%%FORCE_GUARANTEED_ROLLUP%%",
- Boolean.toString(false)
- );
- spec = StringUtils.replace(
- spec,
- "%%SEGMENT_AVAIL_TIMEOUT_MILLIS%%",
- jsonMapper.writeValueAsString("600000")
- );
- return StringUtils.replace(
- spec,
- "%%PARTITIONS_SPEC%%",
- jsonMapper.writeValueAsString(partitionsSpec)
- );
- }
- catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
- };
-
- doIndexTest(
- INDEX_DATASOURCE,
- INDEX_TASK,
- rollupTransform,
- INDEX_QUERIES_RESOURCE,
- false,
- true,
- true,
- new Pair<>(true, true)
- );
- }
- }
-
- /**
- * Test a non zero value for awaitSegmentAvailabilityTimeoutMillis. Setting the config value to 1 millis
- * and pausing coordination to confirm that the task will still succeed even if the job was not able to confirm the
- * segments were loaded by the time the timeout occurs.
- *
- * @param partitionsSpec
- * @throws Exception
- */
- @Test(dataProvider = "resources")
- public void testIndexDataAwaitSegmentAvailabilityFailsButTaskSucceeds(PartitionsSpec partitionsSpec) throws Exception
- {
- try (
- final Closeable ignored1 = unloader(INDEX_DATASOURCE + config.getExtraDatasourceNameSuffix());
- ) {
- coordinatorClient.postDynamicConfig(DYNAMIC_CONFIG_PAUSED);
- boolean forceGuaranteedRollup = partitionsSpec.isForceGuaranteedRollupCompatible();
- Assert.assertFalse(forceGuaranteedRollup, "parititionSpec does not support best-effort rollup");
-
- final Function rollupTransform = spec -> {
- try {
- spec = StringUtils.replace(
- spec,
- "%%FORCE_GUARANTEED_ROLLUP%%",
- Boolean.toString(false)
- );
- spec = StringUtils.replace(
- spec,
- "%%SEGMENT_AVAIL_TIMEOUT_MILLIS%%",
- jsonMapper.writeValueAsString("1")
- );
- return StringUtils.replace(
- spec,
- "%%PARTITIONS_SPEC%%",
- jsonMapper.writeValueAsString(partitionsSpec)
- );
- }
- catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
- };
-
- doIndexTest(
- INDEX_DATASOURCE,
- INDEX_TASK,
- rollupTransform,
- INDEX_QUERIES_RESOURCE,
- false,
- false,
- false,
- new Pair<>(true, false)
- );
- coordinatorClient.postDynamicConfig(DYNAMIC_CONFIG_DEFAULT);
- ITRetryUtil.retryUntilTrue(
- () -> coordinator.areSegmentsLoaded(INDEX_DATASOURCE + config.getExtraDatasourceNameSuffix()), "Segment Load"
- );
- }
- }
-}
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITCompactionTaskTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITCompactionTaskTest.java
deleted file mode 100644
index b974c7d20e97..000000000000
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITCompactionTaskTest.java
+++ /dev/null
@@ -1,405 +0,0 @@
-/*
- * 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.tests.indexer;
-
-import com.google.inject.Inject;
-import org.apache.commons.io.IOUtils;
-import org.apache.druid.indexer.report.IngestionStatsAndErrors;
-import org.apache.druid.indexer.report.IngestionStatsAndErrorsTaskReport;
-import org.apache.druid.indexer.report.TaskReport;
-import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.java.util.common.granularity.GranularityType;
-import org.apache.druid.java.util.common.logger.Logger;
-import org.apache.druid.testing.IntegrationTestingConfig;
-import org.apache.druid.testing.guice.DruidTestModuleFactory;
-import org.apache.druid.testing.utils.ITRetryUtil;
-import org.apache.druid.tests.TestNGGroup;
-import org.joda.time.Interval;
-import org.joda.time.chrono.ISOChronology;
-import org.testng.Assert;
-import org.testng.annotations.BeforeMethod;
-import org.testng.annotations.Guice;
-import org.testng.annotations.Test;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.io.InputStream;
-import java.lang.reflect.Method;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-@Test(groups = {TestNGGroup.COMPACTION, TestNGGroup.QUICKSTART_COMPATIBLE, TestNGGroup.CDS_TASK_SCHEMA_PUBLISH_DISABLED, TestNGGroup.CDS_COORDINATOR_METADATA_QUERY_DISABLED})
-@Guice(moduleFactory = DruidTestModuleFactory.class)
-public class ITCompactionTaskTest extends AbstractIndexerTest
-{
- private static final Logger LOG = new Logger(ITCompactionTaskTest.class);
- private static final String INDEX_TASK = "/indexer/wikipedia_index_task.json";
- private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
-
- private static final String INDEX_QUERIES_YEAR_RESOURCE = "/indexer/wikipedia_index_queries_year_query_granularity.json";
- private static final String INDEX_QUERIES_HOUR_RESOURCE = "/indexer/wikipedia_index_queries_hour_query_granularity.json";
-
- private static final String INDEX_DATASOURCE = "wikipedia_index_test";
-
- private static final String SEGMENT_METADATA_QUERY_RESOURCE = "/indexer/segment_metadata_query.json";
-
- private static final String COMPACTION_TASK = "/indexer/wikipedia_compaction_task.json";
- private static final String PARALLEL_COMPACTION_TASK = "/indexer/wikipedia_compaction_task_parallel.json";
- private static final String COMPACTION_TASK_WITH_SEGMENT_GRANULARITY = "/indexer/wikipedia_compaction_task_with_segment_granularity.json";
- private static final String COMPACTION_TASK_WITH_GRANULARITY_SPEC = "/indexer/wikipedia_compaction_task_with_granularity_spec.json";
-
- private static final String INDEX_TASK_WITH_TIMESTAMP = "/indexer/wikipedia_with_timestamp_index_task.json";
-
- @Inject
- private IntegrationTestingConfig config;
-
- private String fullDatasourceName;
-
- @BeforeMethod
- public void setFullDatasourceName(Method method)
- {
- fullDatasourceName = INDEX_DATASOURCE + config.getExtraDatasourceNameSuffix() + "-" + method.getName();
- }
-
- @Test
- public void testCompaction() throws Exception
- {
- loadDataAndCompact(INDEX_TASK, INDEX_QUERIES_RESOURCE, COMPACTION_TASK, null);
- }
-
- @Test
- public void testCompactionWithSegmentGranularity() throws Exception
- {
- loadDataAndCompact(INDEX_TASK, INDEX_QUERIES_RESOURCE, COMPACTION_TASK_WITH_SEGMENT_GRANULARITY, GranularityType.MONTH);
- }
-
- @Test
- public void testCompactionWithSegmentGranularityInGranularitySpec() throws Exception
- {
- loadDataAndCompact(INDEX_TASK, INDEX_QUERIES_RESOURCE, COMPACTION_TASK_WITH_GRANULARITY_SPEC, GranularityType.MONTH);
- }
-
- @Test
- public void testCompactionWithQueryGranularityInGranularitySpec() throws Exception
- {
- try (final Closeable ignored = unloader(fullDatasourceName)) {
- loadData(INDEX_TASK, fullDatasourceName);
- // 4 segments across 2 days
- checkNumberOfSegments(4);
- List expectedIntervalAfterCompaction = coordinator.getSegmentIntervals(fullDatasourceName);
- expectedIntervalAfterCompaction.sort(null);
-
- checkQueryGranularity(SEGMENT_METADATA_QUERY_RESOURCE, GranularityType.SECOND.name(), 4);
- String queryResponseTemplate = getQueryResponseTemplate(INDEX_QUERIES_RESOURCE);
- queryHelper.testQueriesFromString(queryResponseTemplate);
- // QueryGranularity was SECOND, now we will change it to HOUR (QueryGranularity changed to coarser)
- compactData(COMPACTION_TASK_WITH_GRANULARITY_SPEC, null, GranularityType.HOUR);
-
- // The original 4 segments should be compacted into 2 new segments since data only has 2 days and the compaction
- // segmentGranularity is DAY
- checkNumberOfSegments(2);
- queryResponseTemplate = getQueryResponseTemplate(INDEX_QUERIES_HOUR_RESOURCE);
- queryHelper.testQueriesFromString(queryResponseTemplate);
- checkQueryGranularity(SEGMENT_METADATA_QUERY_RESOURCE, GranularityType.HOUR.name(), 2);
- checkCompactionIntervals(expectedIntervalAfterCompaction);
-
- // QueryGranularity was HOUR, now we will change it to MINUTE (QueryGranularity changed to finer)
- compactData(COMPACTION_TASK_WITH_GRANULARITY_SPEC, null, GranularityType.MINUTE);
-
- // There will be no change in number of segments as compaction segmentGranularity is the same and data interval
- // is the same. Since QueryGranularity is changed to finer qranularity, the data will remains the same. (data
- // will just be bucketed to a finer qranularity but roll up will not be different
- // i.e. 2020-10-29T05:00 will just be bucketed to 2020-10-29T05:00:00)
- checkNumberOfSegments(2);
- queryResponseTemplate = getQueryResponseTemplate(INDEX_QUERIES_HOUR_RESOURCE);
- queryHelper.testQueriesFromString(queryResponseTemplate);
- checkQueryGranularity(SEGMENT_METADATA_QUERY_RESOURCE, GranularityType.MINUTE.name(), 2);
- checkCompactionIntervals(expectedIntervalAfterCompaction);
- }
- }
-
- @Test
- public void testParallelHashedCompaction() throws Exception
- {
- try (final Closeable ignored = unloader(fullDatasourceName)) {
- loadData(INDEX_TASK, fullDatasourceName);
- // 4 segments across 2 days
- checkNumberOfSegments(4);
- List expectedIntervalAfterCompaction = coordinator.getSegmentIntervals(fullDatasourceName);
- expectedIntervalAfterCompaction.sort(null);
-
- checkQueryGranularity(SEGMENT_METADATA_QUERY_RESOURCE, GranularityType.SECOND.name(), 4);
- String queryResponseTemplate = getQueryResponseTemplate(INDEX_QUERIES_RESOURCE);
-
- queryResponseTemplate = StringUtils.replace(
- queryResponseTemplate,
- "%%SEGMENT_AVAIL_TIMEOUT_MILLIS%%",
- jsonMapper.writeValueAsString("0")
- );
-
- queryHelper.testQueriesFromString(queryResponseTemplate);
- String taskId = compactData(PARALLEL_COMPACTION_TASK, null, null);
-
- // The original 4 segments should be compacted into 2 new segments
- checkNumberOfSegments(2);
- queryHelper.testQueriesFromString(queryResponseTemplate);
- checkQueryGranularity(SEGMENT_METADATA_QUERY_RESOURCE, GranularityType.SECOND.name(), 2);
-
-
- checkCompactionIntervals(expectedIntervalAfterCompaction);
-
- Map reports = indexer.getTaskReport(taskId);
- Assert.assertTrue(reports != null && reports.size() > 0);
-
- Assert.assertEquals(
- 2,
- reports.values()
- .stream()
- .filter(r -> r instanceof IngestionStatsAndErrorsTaskReport)
- .mapToLong(r -> ((IngestionStatsAndErrors) r.getPayload()).getSegmentsPublished())
- .sum()
- );
- Assert.assertEquals(
- 4,
- reports.values()
- .stream()
- .filter(r -> r instanceof IngestionStatsAndErrorsTaskReport)
- .mapToLong(r -> ((IngestionStatsAndErrors) r.getPayload()).getSegmentsRead())
- .sum()
- );
- }
- }
-
- @Test
- public void testCompactionWithSegmentGranularityAndQueryGranularityInGranularitySpec() throws Exception
- {
- try (final Closeable ignored = unloader(fullDatasourceName)) {
- loadData(INDEX_TASK, fullDatasourceName);
- // 4 segments across 2 days
- checkNumberOfSegments(4);
- List expectedIntervalAfterCompaction = coordinator.getSegmentIntervals(fullDatasourceName);
- expectedIntervalAfterCompaction.sort(null);
-
- checkQueryGranularity(SEGMENT_METADATA_QUERY_RESOURCE, GranularityType.SECOND.name(), 4);
- String queryResponseTemplate = getQueryResponseTemplate(INDEX_QUERIES_RESOURCE);
- queryHelper.testQueriesFromString(queryResponseTemplate);
- compactData(COMPACTION_TASK_WITH_GRANULARITY_SPEC, GranularityType.YEAR, GranularityType.YEAR);
-
- // The original 4 segments should be compacted into 1 new segment
- checkNumberOfSegments(1);
- queryResponseTemplate = getQueryResponseTemplate(INDEX_QUERIES_YEAR_RESOURCE);
- queryHelper.testQueriesFromString(queryResponseTemplate);
- checkQueryGranularity(SEGMENT_METADATA_QUERY_RESOURCE, GranularityType.YEAR.name(), 1);
-
- List newIntervals = new ArrayList<>();
- for (String interval : expectedIntervalAfterCompaction) {
- for (Interval newinterval : GranularityType.YEAR.getDefaultGranularity().getIterable(new Interval(interval, ISOChronology.getInstanceUTC()))) {
- newIntervals.add(newinterval.toString());
- }
- }
- expectedIntervalAfterCompaction = newIntervals;
- checkCompactionIntervals(expectedIntervalAfterCompaction);
- }
- }
-
- @Test
- public void testCompactionWithTimestampDimension() throws Exception
- {
- loadDataAndCompact(INDEX_TASK_WITH_TIMESTAMP, INDEX_QUERIES_RESOURCE, COMPACTION_TASK, null);
- }
-
- private void loadDataAndCompact(
- String indexTask,
- String queriesResource,
- String compactionResource,
- GranularityType newSegmentGranularity
- ) throws Exception
- {
- try (final Closeable ignored = unloader(fullDatasourceName)) {
- loadData(indexTask, fullDatasourceName);
- // 4 segments across 2 days
- checkNumberOfSegments(4);
- List expectedIntervalAfterCompaction = coordinator.getSegmentIntervals(fullDatasourceName);
- expectedIntervalAfterCompaction.sort(null);
-
- checkQueryGranularity(SEGMENT_METADATA_QUERY_RESOURCE, GranularityType.SECOND.name(), 4);
- String queryResponseTemplate = getQueryResponseTemplate(queriesResource);
-
- queryResponseTemplate = StringUtils.replace(
- queryResponseTemplate,
- "%%SEGMENT_AVAIL_TIMEOUT_MILLIS%%",
- jsonMapper.writeValueAsString("0")
- );
-
- queryHelper.testQueriesFromString(queryResponseTemplate);
- String taskId = compactData(compactionResource, newSegmentGranularity, null);
-
- // The original 4 segments should be compacted into 2 new segments
- checkNumberOfSegments(2);
- queryHelper.testQueriesFromString(queryResponseTemplate);
- checkQueryGranularity(SEGMENT_METADATA_QUERY_RESOURCE, GranularityType.SECOND.name(), 2);
-
- if (newSegmentGranularity != null) {
- List newIntervals = new ArrayList<>();
- for (String interval : expectedIntervalAfterCompaction) {
- for (Interval newinterval : newSegmentGranularity.getDefaultGranularity().getIterable(new Interval(interval, ISOChronology.getInstanceUTC()))) {
- newIntervals.add(newinterval.toString());
- }
- }
- expectedIntervalAfterCompaction = newIntervals;
- }
- checkCompactionIntervals(expectedIntervalAfterCompaction);
-
- Map reports = indexer.getTaskReport(taskId);
- Assert.assertTrue(reports != null && reports.size() > 0);
- }
- }
-
- private String compactData(
- String compactionResource,
- GranularityType newSegmentGranularity,
- GranularityType newQueryGranularity
- ) throws Exception
- {
- String template = getResourceAsString(compactionResource);
- template = StringUtils.replace(template, "%%DATASOURCE%%", fullDatasourceName);
- // For the new granularitySpec map
- Map granularityMap = new HashMap<>();
- if (newSegmentGranularity != null) {
- granularityMap.put("segmentGranularity", newSegmentGranularity.name());
- }
- if (newQueryGranularity != null) {
- granularityMap.put("queryGranularity", newQueryGranularity.name());
- }
- template = StringUtils.replace(
- template,
- "%%GRANULARITY_SPEC%%",
- jsonMapper.writeValueAsString(granularityMap)
- );
- // For the deprecated segment granularity field
- if (newSegmentGranularity != null) {
- template = StringUtils.replace(
- template,
- "%%SEGMENT_GRANULARITY%%",
- newSegmentGranularity.name()
- );
- }
- final String taskID = indexer.submitTask(template);
- LOG.info("TaskID for compaction task %s", taskID);
- indexer.waitUntilTaskCompletes(taskID);
-
- ITRetryUtil.retryUntilTrue(
- () -> coordinator.areSegmentsLoaded(fullDatasourceName),
- "Segment Compaction"
- );
-
- return taskID;
- }
-
- private void checkQueryGranularity(String queryResource, String expectedQueryGranularity, int segmentCount) throws Exception
- {
- String queryResponseTemplate;
- try {
- InputStream is = AbstractITBatchIndexTest.class.getResourceAsStream(queryResource);
- queryResponseTemplate = IOUtils.toString(is, StandardCharsets.UTF_8);
- }
- catch (IOException e) {
- throw new ISE(e, "could not read query file: %s", queryResource);
- }
-
- queryResponseTemplate = StringUtils.replace(
- queryResponseTemplate,
- "%%DATASOURCE%%",
- fullDatasourceName
- );
- queryResponseTemplate = StringUtils.replace(
- queryResponseTemplate,
- "%%ANALYSIS_TYPE%%",
- "queryGranularity"
- );
- queryResponseTemplate = StringUtils.replace(
- queryResponseTemplate,
- "%%INTERVALS%%",
- "2013-08-31/2013-09-02"
- );
- List> expectedResults = new ArrayList<>();
- for (int i = 0; i < segmentCount; i++) {
- Map result = new HashMap<>();
- result.put("queryGranularity", expectedQueryGranularity);
- expectedResults.add(result);
- }
- queryResponseTemplate = StringUtils.replace(
- queryResponseTemplate,
- "%%EXPECTED_QUERY_GRANULARITY%%",
- jsonMapper.writeValueAsString(expectedResults)
- );
- queryHelper.testQueriesFromString(queryResponseTemplate);
- }
-
- private void checkNumberOfSegments(int numExpectedSegments)
- {
- ITRetryUtil.retryUntilTrue(
- () -> {
- int metadataSegmentCount = coordinator.getSegments(fullDatasourceName).size();
- LOG.info("Current metadata segment count: %d, expected: %d", metadataSegmentCount, numExpectedSegments);
- return metadataSegmentCount == numExpectedSegments;
- },
- "Segment count check"
- );
- }
-
- private void checkCompactionIntervals(List expectedIntervals)
- {
- Set expectedIntervalsSet = new HashSet<>(expectedIntervals);
- ITRetryUtil.retryUntilTrue(
- () -> {
- final Set intervalsAfterCompaction = new HashSet<>(coordinator.getSegmentIntervals(fullDatasourceName));
- System.out.println("ACTUAL: " + intervalsAfterCompaction);
- System.out.println("EXPECTED: " + expectedIntervalsSet);
- return intervalsAfterCompaction.equals(expectedIntervalsSet);
- },
- "Compaction interval check"
- );
- }
-
- private String getQueryResponseTemplate(String queryResourcePath)
- {
- String queryResponseTemplate;
- try {
- InputStream is = AbstractITBatchIndexTest.class.getResourceAsStream(queryResourcePath);
- queryResponseTemplate = IOUtils.toString(is, StandardCharsets.UTF_8);
- queryResponseTemplate = StringUtils.replace(
- queryResponseTemplate,
- "%%DATASOURCE%%",
- fullDatasourceName
- );
- }
- catch (IOException e) {
- throw new ISE(e, "could not read query file: %s", queryResourcePath);
- }
- return queryResponseTemplate;
- }
-}
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITPerfectRollupParallelIndexTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITPerfectRollupParallelIndexTest.java
deleted file mode 100644
index 78823477434b..000000000000
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITPerfectRollupParallelIndexTest.java
+++ /dev/null
@@ -1,129 +0,0 @@
-/*
- * 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.tests.indexer;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
-import org.apache.druid.indexer.partitions.PartitionsSpec;
-import org.apache.druid.indexer.partitions.SingleDimensionPartitionsSpec;
-import org.apache.druid.java.util.common.Pair;
-import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.testing.guice.DruidTestModuleFactory;
-import org.apache.druid.tests.TestNGGroup;
-import org.apache.druid.timeline.partition.HashPartitionFunction;
-import org.testng.Assert;
-import org.testng.annotations.DataProvider;
-import org.testng.annotations.Guice;
-import org.testng.annotations.Test;
-
-import java.io.Closeable;
-import java.util.function.Function;
-
-@Test(groups = {TestNGGroup.PERFECT_ROLLUP_PARALLEL_BATCH_INDEX, TestNGGroup.SHUFFLE_DEEP_STORE})
-@Guice(moduleFactory = DruidTestModuleFactory.class)
-public class ITPerfectRollupParallelIndexTest extends AbstractITBatchIndexTest
-{
- // This ingestion spec has a splitHintSpec of maxSplitSize of 1 to test whether or not the task can handle
- // maxSplitSize of 1 properly.
- private static final String INDEX_TASK = "/indexer/wikipedia_parallel_index_task.json";
- private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_parallel_index_queries.json";
- private static final String INDEX_DATASOURCE = "wikipedia_parallel_index_test";
- private static final String INDEX_INGEST_SEGMENT_DATASOURCE = "wikipedia_parallel_ingest_segment_index_test";
- private static final String INDEX_INGEST_SEGMENT_TASK = "/indexer/wikipedia_parallel_ingest_segment_index_task.json";
- private static final String INDEX_DRUID_INPUT_SOURCE_DATASOURCE = "wikipedia_parallel_druid_input_source_index_test";
- private static final String INDEX_DRUID_INPUT_SOURCE_TASK = "/indexer/wikipedia_parallel_druid_input_source_index_task.json";
-
- @DataProvider
- public static Object[][] resources()
- {
- return new Object[][]{
- {new HashedPartitionsSpec(null, 2, null, HashPartitionFunction.MURMUR3_32_ABS)},
- {new HashedPartitionsSpec(null, 2, null, null)},
- {new SingleDimensionPartitionsSpec(2, null, "namespace", false)}
- };
- }
-
- @Test(dataProvider = "resources")
- public void testIndexData(PartitionsSpec partitionsSpec) throws Exception
- {
- try (
- final Closeable ignored1 = unloader(INDEX_DATASOURCE + config.getExtraDatasourceNameSuffix());
- final Closeable ignored2 = unloader(INDEX_INGEST_SEGMENT_DATASOURCE + config.getExtraDatasourceNameSuffix());
- final Closeable ignored3 = unloader(INDEX_DRUID_INPUT_SOURCE_DATASOURCE + config.getExtraDatasourceNameSuffix())
- ) {
- boolean forceGuaranteedRollup = partitionsSpec.isForceGuaranteedRollupCompatible();
- Assert.assertTrue(forceGuaranteedRollup, "parititionSpec does not support perfect rollup");
-
- final Function rollupTransform = spec -> {
- try {
- spec = StringUtils.replace(
- spec,
- "%%FORCE_GUARANTEED_ROLLUP%%",
- Boolean.toString(true)
- );
- spec = StringUtils.replace(
- spec,
- "%%SEGMENT_AVAIL_TIMEOUT_MILLIS%%",
- jsonMapper.writeValueAsString("0")
- );
- return StringUtils.replace(
- spec,
- "%%PARTITIONS_SPEC%%",
- jsonMapper.writeValueAsString(partitionsSpec)
- );
- }
- catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
- };
-
- doIndexTest(
- INDEX_DATASOURCE,
- INDEX_TASK,
- rollupTransform,
- INDEX_QUERIES_RESOURCE,
- false,
- true,
- true,
- new Pair<>(false, false)
- );
-
- doReindexTest(
- INDEX_DATASOURCE,
- INDEX_INGEST_SEGMENT_DATASOURCE,
- rollupTransform,
- INDEX_INGEST_SEGMENT_TASK,
- INDEX_QUERIES_RESOURCE,
- new Pair<>(false, false)
-
- );
-
- // with DruidInputSource
- doReindexTest(
- INDEX_DATASOURCE,
- INDEX_DRUID_INPUT_SOURCE_DATASOURCE,
- rollupTransform,
- INDEX_DRUID_INPUT_SOURCE_TASK,
- INDEX_QUERIES_RESOURCE,
- new Pair<>(false, false)
- );
- }
- }
-}
diff --git a/integration-tests/src/test/resources/api/overlord-resource-test-task.json b/integration-tests/src/test/resources/api/overlord-resource-test-task.json
deleted file mode 100644
index a3f99ad62543..000000000000
--- a/integration-tests/src/test/resources/api/overlord-resource-test-task.json
+++ /dev/null
@@ -1,87 +0,0 @@
-{
- "type": "index_parallel",
- "spec": {
- "dataSchema": {
- "dataSource": "it-overlord-resource-test",
- "timestampSpec": {
- "column": "timestamp"
- },
- "dimensionsSpec": {
- "dimensions": [
- "page",
- {"type": "string", "name": "language", "createBitmapIndex": false},
- "user",
- "unpatrolled",
- "newPage",
- "robot",
- "anonymous",
- "namespace",
- "continent",
- "country",
- "region",
- "city"
- ]
- },
- "metricsSpec": [
- {
- "type": "count",
- "name": "count"
- },
- {
- "type": "doubleSum",
- "name": "added",
- "fieldName": "added"
- },
- {
- "type": "doubleSum",
- "name": "deleted",
- "fieldName": "deleted"
- },
- {
- "type": "doubleSum",
- "name": "delta",
- "fieldName": "delta"
- },
- {
- "name": "thetaSketch",
- "type": "thetaSketch",
- "fieldName": "user"
- },
- {
- "name": "quantilesDoublesSketch",
- "type": "quantilesDoublesSketch",
- "fieldName": "delta"
- },
- {
- "name": "HLLSketchBuild",
- "type": "HLLSketchBuild",
- "fieldName": "user"
- }
- ],
- "granularitySpec": {
- "segmentGranularity": "DAY",
- "queryGranularity": "second",
- "intervals" : [ "2013-08-31/2013-09-02" ]
- }
- },
- "ioConfig": {
- "type": "index_parallel",
- "inputSource": {
- "type": "local",
- "filter" : "*.json",
- "baseDir": "/resources/data/batch_index/json"
- },
- "inputFormat": {
- "type": "json"
- }
- },
- "tuningConfig": {
- "type": "index_parallel",
- "maxNumConcurrentSubTasks": 1,
- "splitHintSpec": {
- "type": "maxSize",
- "maxNumFiles": 1
- }
- }
- }
-}
\ No newline at end of file
diff --git a/integration-tests/src/test/resources/indexer/sparse_column_index_queries.json b/integration-tests/src/test/resources/indexer/sparse_column_index_queries.json
deleted file mode 100644
index 193e69fbc9ca..000000000000
--- a/integration-tests/src/test/resources/indexer/sparse_column_index_queries.json
+++ /dev/null
@@ -1,80 +0,0 @@
-[
- {
- "description": "timeseries, 1 agg, all",
- "query":{
- "queryType" : "timeBoundary",
- "dataSource": "%%DATASOURCE%%"
- },
- "expectedResults":[
- {
- "timestamp" : "2015-09-12T00:00:00.000Z",
- "result" : {
- "minTime" : "2015-09-12T00:00:00.000Z",
- "maxTime" : "2015-09-12T00:00:00.000Z"
- }
- }
- ]
- },
- {
- "description": "scan, all",
- "query": {
- "queryType": "scan",
- "dataSource": "%%DATASOURCE%%",
- "intervals": [
- "2013-01-01/2020-01-02"
- ],
- "resultFormat":"compactedList"
- },
- "expectedResults": %%EXPECTED_SCAN_RESULT%%,
- "fieldsToTest": ["events"]
- },
- {
- "description": "roll up ratio",
- "query": {
- "queryType":"timeseries",
- "dataSource":{
- "type":"table",
- "name":"%%DATASOURCE%%"
- },
- "intervals":{
- "type":"intervals",
- "intervals":[
- "2013-01-01/2020-01-02"
- ]
- },
- "granularity":{
- "type":"all"
- },
- "aggregations":[
- {
- "type":"count",
- "name":"a0"
- },
- {
- "type":"longSum",
- "name":"a1",
- "fieldName":"count",
- "expression":null
- }
- ],
- "postAggregations":[
- {
- "type":"expression",
- "name":"p0",
- "expression":"((\"a0\" * 1.00) / \"a1\")",
- "ordering":null
- }
- ]
- },
- "expectedResults": [
- {
- "timestamp" : "2015-09-12T00:00:00.000Z",
- "result" : {
- "a1" : %%EXPECTED_SUM_COUNT%%,
- "p0" : %%EXPECTED_ROLLUP_RATIO%%,
- "a0" : %%EXPECTED_NUM_ROW%%
- }
- }
- ]
- }
-]
\ No newline at end of file
diff --git a/integration-tests/src/test/resources/indexer/sparse_column_index_task.json b/integration-tests/src/test/resources/indexer/sparse_column_index_task.json
deleted file mode 100644
index 3a21a856ac6f..000000000000
--- a/integration-tests/src/test/resources/indexer/sparse_column_index_task.json
+++ /dev/null
@@ -1,57 +0,0 @@
-{
- "type": "index_parallel",
- "spec": {
- "ioConfig": {
- "type": "index_parallel",
- "inputSource": {
- "type": "inline",
- "data": "{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"C\",\"dimB\":\"F\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"C\",\"dimB\":\"J\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"H\",\"dimB\":\"X\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"Z\",\"dimB\":\"S\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"H\",\"dimB\":\"X\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"H\",\"dimB\":\"Z\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"J\",\"dimB\":\"R\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"H\",\"dimB\":\"T\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimA\":\"H\",\"dimB\":\"X\",\"metA\":1}\n{\"time\":\"2015-09-12T00:46:58.771Z\",\"dimC\":\"A\",\"dimB\":\"X\",\"metA\":1}\n"
- },
- "inputFormat": {
- "type": "json"
- }
- },
- "tuningConfig": {
- "type": "index_parallel",
- "partitionsSpec": {
- "type": "dynamic",
- "maxRowsPerSegment": 3,
- "maxTotalRows": 3
- },
- "maxRowsInMemory": 3
- },
- "dataSchema": {
- "dataSource": "%%DATASOURCE%%",
- "timestampSpec": {
- "column": "time",
- "format": "iso"
- },
- "dimensionsSpec": {
- "dimensions": [
- "dimB",
- "dimA",
- "dimC",
- "dimD",
- "dimE",
- "dimF"
- ]
- },
- "granularitySpec": {
- "queryGranularity": "hour",
- "rollup": true,
- "segmentGranularity": "hour"
- },
- "metricsSpec": [
- {
- "name": "count",
- "type": "count"
- },
- {
- "name": "sum_metA",
- "type": "longSum",
- "fieldName": "metA"
- }
- ]
- }
- }
-}
\ No newline at end of file
diff --git a/integration-tests/src/test/resources/indexer/sparse_column_with_dim_compaction_task.json b/integration-tests/src/test/resources/indexer/sparse_column_with_dim_compaction_task.json
deleted file mode 100644
index 9416a3f6bdad..000000000000
--- a/integration-tests/src/test/resources/indexer/sparse_column_with_dim_compaction_task.json
+++ /dev/null
@@ -1,19 +0,0 @@
-{
- "type": "compact",
- "dataSource": "%%DATASOURCE%%",
- "dimensionsSpec": {
- "dimensions": %%DIMENSION_NAMES%%
- },
- "interval": "2010-10-29T05:00:00Z/2030-10-29T06:00:00Z",
- "tuningConfig": {
- "type": "index_parallel",
- "maxRowsPerSegment": 3,
- "maxRowsInMemory": 3,
- "maxNumConcurrentSubTasks": 2,
- "partitionsSpec": {
- "type": "hashed",
- "numShards": 1
- },
- "forceGuaranteedRollup": true
- }
-}
\ No newline at end of file
diff --git a/integration-tests/src/test/resources/indexer/sparse_column_without_dim_compaction_task.json b/integration-tests/src/test/resources/indexer/sparse_column_without_dim_compaction_task.json
deleted file mode 100644
index a149d7a25127..000000000000
--- a/integration-tests/src/test/resources/indexer/sparse_column_without_dim_compaction_task.json
+++ /dev/null
@@ -1,16 +0,0 @@
-{
- "type": "compact",
- "dataSource": "%%DATASOURCE%%",
- "interval": "2010-10-29T05:00:00Z/2030-10-29T06:00:00Z",
- "tuningConfig": {
- "type": "index_parallel",
- "maxRowsPerSegment": 3,
- "maxRowsInMemory": 3,
- "maxNumConcurrentSubTasks": 2,
- "partitionsSpec": {
- "type": "hashed",
- "numShards": 1
- },
- "forceGuaranteedRollup": true
- }
-}
\ No newline at end of file
diff --git a/integration-tests/src/test/resources/indexer/wikipedia_index_queries_hour_query_granularity.json b/integration-tests/src/test/resources/indexer/wikipedia_index_queries_hour_query_granularity.json
deleted file mode 100644
index 6ed715b89eb5..000000000000
--- a/integration-tests/src/test/resources/indexer/wikipedia_index_queries_hour_query_granularity.json
+++ /dev/null
@@ -1,150 +0,0 @@
-[
- {
- "description": "timeseries, 1 agg, all",
- "query":{
- "queryType" : "timeBoundary",
- "dataSource": "%%DATASOURCE%%"
- },
- "expectedResults":[
- {
- "timestamp" : "2013-08-31T01:00:00.000Z",
- "result" : {
- "minTime" : "2013-08-31T01:00:00.000Z",
- "maxTime" : "2013-09-01T12:00:00.000Z"
- }
- }
- ]
- },
- {
- "description": "timeseries, datasketch aggs, all",
- "query":{
- "queryType" : "timeseries",
- "dataSource": "%%DATASOURCE%%",
- "granularity":"day",
- "intervals":[
- "2013-08-31T00:00/2013-09-01T00:00"
- ],
- "filter":null,
- "aggregations":[
- {
- "type": "HLLSketchMerge",
- "name": "approxCountHLL",
- "fieldName": "HLLSketchBuild",
- "lgK": 12,
- "tgtHllType": "HLL_4",
- "round": true
- },
- {
- "type":"thetaSketch",
- "name":"approxCountTheta",
- "fieldName":"thetaSketch",
- "size":16384,
- "shouldFinalize":true,
- "isInputThetaSketch":false,
- "errorBoundsStdDev":null
- },
- {
- "type":"quantilesDoublesSketch",
- "name":"quantilesSketch",
- "fieldName":"quantilesDoublesSketch",
- "k":128
- }
- ]
- },
- "expectedResults":[
- {
- "timestamp" : "2013-08-31T00:00:00.000Z",
- "result" : {
- "quantilesSketch":5,
- "approxCountTheta":5.0,
- "approxCountHLL":5
- }
- }
- ]
- },
- {
- "description":"having spec on post aggregation",
- "query":{
- "queryType":"groupBy",
- "dataSource":"%%DATASOURCE%%",
- "granularity":"day",
- "dimensions":[
- "page"
- ],
- "filter":{
- "type":"selector",
- "dimension":"language",
- "value":"zh"
- },
- "aggregations":[
- {
- "type":"count",
- "name":"rows"
- },
- {
- "type":"longSum",
- "fieldName":"added",
- "name":"added_count"
- }
- ],
- "postAggregations": [
- {
- "type":"arithmetic",
- "name":"added_count_times_ten",
- "fn":"*",
- "fields":[
- {"type":"fieldAccess", "name":"added_count", "fieldName":"added_count"},
- {"type":"constant", "name":"const", "value":10}
- ]
- }
- ],
- "having":{"type":"greaterThan", "aggregation":"added_count_times_ten", "value":9000},
- "intervals":[
- "2013-08-31T00:00/2013-09-01T00:00"
- ]
- },
- "expectedResults":[ {
- "version" : "v1",
- "timestamp" : "2013-08-31T00:00:00.000Z",
- "event" : {
- "added_count_times_ten" : 9050.0,
- "page" : "Crimson Typhoon",
- "added_count" : 905,
- "rows" : 1
- }
- } ]
- },
- {
- "description": "timeseries, stringFirst/stringLast aggs, all",
- "query":{
- "queryType" : "timeseries",
- "dataSource": "%%DATASOURCE%%",
- "granularity":"day",
- "intervals":[
- "2013-08-31T00:00/2013-09-01T00:00"
- ],
- "filter":null,
- "aggregations":[
- {
- "type": "stringFirst",
- "name": "first_user",
- "fieldName": "user"
- },
- {
- "type":"stringLast",
- "name":"last_user",
- "fieldName":"user"
- }
- ]
- },
- "expectedResults":[
- {
- "timestamp" : "2013-08-31T00:00:00.000Z",
- "result" : {
- "first_user":"nuclear",
- "last_user":"stringer"
- }
- }
- ]
- }
-]
\ No newline at end of file
diff --git a/integration-tests/src/test/resources/indexer/wikipedia_index_queries_year_query_granularity.json b/integration-tests/src/test/resources/indexer/wikipedia_index_queries_year_query_granularity.json
deleted file mode 100644
index 25f9203ac1b8..000000000000
--- a/integration-tests/src/test/resources/indexer/wikipedia_index_queries_year_query_granularity.json
+++ /dev/null
@@ -1,150 +0,0 @@
-[
- {
- "description": "timeseries, 1 agg, all",
- "query":{
- "queryType" : "timeBoundary",
- "dataSource": "%%DATASOURCE%%"
- },
- "expectedResults":[
- {
- "timestamp" : "2013-01-01T00:00:00.000Z",
- "result" : {
- "minTime" : "2013-01-01T00:00:00.000Z",
- "maxTime" : "2013-01-01T00:00:00.000Z"
- }
- }
- ]
- },
- {
- "description": "timeseries, datasketch aggs, all",
- "query":{
- "queryType" : "timeseries",
- "dataSource": "%%DATASOURCE%%",
- "granularity":"day",
- "intervals":[
- "2013-01-01T00:00/2014-01-01T00:00"
- ],
- "filter":null,
- "aggregations":[
- {
- "type": "HLLSketchMerge",
- "name": "approxCountHLL",
- "fieldName": "HLLSketchBuild",
- "lgK": 12,
- "tgtHllType": "HLL_4",
- "round": true
- },
- {
- "type":"thetaSketch",
- "name":"approxCountTheta",
- "fieldName":"thetaSketch",
- "size":16384,
- "shouldFinalize":true,
- "isInputThetaSketch":false,
- "errorBoundsStdDev":null
- },
- {
- "type":"quantilesDoublesSketch",
- "name":"quantilesSketch",
- "fieldName":"quantilesDoublesSketch",
- "k":128
- }
- ]
- },
- "expectedResults":[
- {
- "timestamp" : "2013-01-01T00:00:00.000Z",
- "result" : {
- "quantilesSketch":10,
- "approxCountTheta":5.0,
- "approxCountHLL":5
- }
- }
- ]
- },
- {
- "description":"having spec on post aggregation",
- "query":{
- "queryType":"groupBy",
- "dataSource":"%%DATASOURCE%%",
- "granularity":"day",
- "dimensions":[
- "page"
- ],
- "filter":{
- "type":"selector",
- "dimension":"language",
- "value":"zh"
- },
- "aggregations":[
- {
- "type":"count",
- "name":"rows"
- },
- {
- "type":"longSum",
- "fieldName":"added",
- "name":"added_count"
- }
- ],
- "postAggregations": [
- {
- "type":"arithmetic",
- "name":"added_count_times_ten",
- "fn":"*",
- "fields":[
- {"type":"fieldAccess", "name":"added_count", "fieldName":"added_count"},
- {"type":"constant", "name":"const", "value":10}
- ]
- }
- ],
- "having":{"type":"greaterThan", "aggregation":"added_count_times_ten", "value":9000},
- "intervals":[
- "2013-01-01T00:00/2014-01-01T00:00"
- ]
- },
- "expectedResults":[ {
- "version" : "v1",
- "timestamp" : "2013-01-01T00:00:00.000Z",
- "event" : {
- "added_count_times_ten" : 18100.0,
- "page" : "Crimson Typhoon",
- "added_count" : 1810,
- "rows" : 1
- }
- } ]
- },
- {
- "description": "timeseries, stringFirst/stringLast aggs, all",
- "query":{
- "queryType" : "timeseries",
- "dataSource": "%%DATASOURCE%%",
- "granularity":"day",
- "intervals":[
- "2013-01-01T00:00/2014-01-01T00:00"
- ],
- "filter":null,
- "aggregations":[
- {
- "type": "stringFirst",
- "name": "first_user",
- "fieldName": "user"
- },
- {
- "type":"stringLast",
- "name":"last_user",
- "fieldName":"user"
- }
- ]
- },
- "expectedResults":[
- {
- "timestamp" : "2013-01-01T00:00:00.000Z",
- "result" : {
- "first_user":"masterYi",
- "last_user":"speed"
- }
- }
- ]
- }
-]
\ No newline at end of file
diff --git a/processing/src/main/java/org/apache/druid/indexer/report/IngestionStatsAndErrors.java b/processing/src/main/java/org/apache/druid/indexer/report/IngestionStatsAndErrors.java
index fe6566947889..f766d0a324cb 100644
--- a/processing/src/main/java/org/apache/druid/indexer/report/IngestionStatsAndErrors.java
+++ b/processing/src/main/java/org/apache/druid/indexer/report/IngestionStatsAndErrors.java
@@ -175,7 +175,7 @@ public String toString()
", unparseableEvents=" + unparseableEvents +
", rowStats=" + rowStats +
", errorMsg='" + errorMsg + '\'' +
- ", segmentAvailabilityConfoirmed=" + segmentAvailabilityConfirmed +
+ ", segmentAvailabilityConfirmed=" + segmentAvailabilityConfirmed +
", segmentAvailabilityWaitTimeMs=" + segmentAvailabilityWaitTimeMs +
", recordsProcessed=" + recordsProcessed +
", segmentsRead=" + segmentsRead +
diff --git a/server/src/main/java/org/apache/druid/client/broker/BrokerClient.java b/server/src/main/java/org/apache/druid/client/broker/BrokerClient.java
index 30bfa22937d9..377866c48ee4 100644
--- a/server/src/main/java/org/apache/druid/client/broker/BrokerClient.java
+++ b/server/src/main/java/org/apache/druid/client/broker/BrokerClient.java
@@ -20,6 +20,7 @@
package org.apache.druid.client.broker;
import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.druid.query.Query;
import org.apache.druid.query.explain.ExplainPlan;
import org.apache.druid.query.http.ClientSqlQuery;
import org.apache.druid.query.http.SqlTaskStatus;
@@ -38,6 +39,11 @@
*/
public interface BrokerClient
{
+ /**
+ * Submits the given query to the {@code /druid/v2} endpoint of a Broker.
+ */
+ ListenableFuture submitNativeQuery(Query query);
+
/**
* Submit the given {@code sqlQuery} to the Broker's SQL query endpoint, {@code /druid/v2/sql/}.
*/
diff --git a/server/src/main/java/org/apache/druid/client/broker/BrokerClientImpl.java b/server/src/main/java/org/apache/druid/client/broker/BrokerClientImpl.java
index 3bce514c9801..52f7c3184d56 100644
--- a/server/src/main/java/org/apache/druid/client/broker/BrokerClientImpl.java
+++ b/server/src/main/java/org/apache/druid/client/broker/BrokerClientImpl.java
@@ -28,6 +28,7 @@
import org.apache.druid.java.util.http.client.response.BytesFullResponseHandler;
import org.apache.druid.java.util.http.client.response.FullResponseHolder;
import org.apache.druid.java.util.http.client.response.StringFullResponseHandler;
+import org.apache.druid.query.Query;
import org.apache.druid.query.explain.ExplainPlan;
import org.apache.druid.query.http.ClientSqlQuery;
import org.apache.druid.query.http.SqlTaskStatus;
@@ -51,6 +52,19 @@ public BrokerClientImpl(final ServiceClient client, final ObjectMapper jsonMappe
this.jsonMapper = jsonMapper;
}
+ @Override
+ public ListenableFuture submitNativeQuery(Query query)
+ {
+ return FutureUtils.transform(
+ client.asyncRequest(
+ new RequestBuilder(HttpMethod.POST, "/druid/v2")
+ .jsonContent(jsonMapper, query),
+ new StringFullResponseHandler(StandardCharsets.UTF_8)
+ ),
+ FullResponseHolder::getContent
+ );
+ }
+
@Override
public ListenableFuture submitSqlQuery(final ClientSqlQuery sqlQuery)
{
diff --git a/server/src/test/java/org/apache/druid/client/broker/BrokerClientImplTest.java b/server/src/test/java/org/apache/druid/client/broker/BrokerClientImplTest.java
index 90d1b49534d3..782191732e26 100644
--- a/server/src/test/java/org/apache/druid/client/broker/BrokerClientImplTest.java
+++ b/server/src/test/java/org/apache/druid/client/broker/BrokerClientImplTest.java
@@ -26,12 +26,15 @@
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.query.Druids;
import org.apache.druid.query.explain.ExplainAttributes;
import org.apache.druid.query.explain.ExplainPlan;
import org.apache.druid.query.http.ClientSqlQuery;
import org.apache.druid.query.http.SqlTaskStatus;
+import org.apache.druid.query.scan.ScanQuery;
import org.apache.druid.rpc.MockServiceClient;
import org.apache.druid.rpc.RequestBuilder;
+import org.apache.druid.segment.TestDataSource;
import org.jboss.netty.handler.codec.http.HttpMethod;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import org.junit.After;
@@ -65,6 +68,31 @@ public void tearDown()
serviceClient.verify();
}
+ @Test
+ public void test_submitNativeQuery() throws Exception
+ {
+ final ScanQuery scanQuery = Druids
+ .newScanQueryBuilder()
+ .dataSource(TestDataSource.WIKI)
+ .eternityInterval()
+ .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_COMPACTED_LIST)
+ .build();
+
+ final Map response = Map.of("events", List.of());
+ serviceClient.expectAndRespond(
+ new RequestBuilder(HttpMethod.POST, "/druid/v2")
+ .jsonContent(jsonMapper, scanQuery),
+ HttpResponseStatus.OK,
+ Map.of(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON),
+ jsonMapper.writeValueAsBytes(response)
+ );
+
+ assertEquals(
+ jsonMapper.writeValueAsString(response),
+ brokerClient.submitNativeQuery(scanQuery).get()
+ );
+ }
+
@Test
public void testSubmitSqlTask() throws Exception
{
diff --git a/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java b/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java
index 6e531848f9e5..f8bb3d66cb8d 100644
--- a/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java
+++ b/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java
@@ -28,19 +28,30 @@
import org.apache.druid.common.utils.IdUtils;
import org.apache.druid.indexer.TaskStatus;
import org.apache.druid.indexing.common.task.TaskMetrics;
+import org.apache.druid.indexing.overlord.Segments;
import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Comparators;
import org.apache.druid.query.DruidMetrics;
import org.apache.druid.query.http.ClientSqlQuery;
import org.apache.druid.rpc.indexing.OverlordClient;
import org.apache.druid.segment.TestDataSource;
import org.apache.druid.segment.TestHelper;
import org.apache.druid.sql.http.ResultFormat;
+import org.apache.druid.timeline.DataSegment;
+import org.joda.time.Interval;
+import org.joda.time.chrono.ISOChronology;
import org.junit.jupiter.api.Assertions;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
import java.util.function.Function;
/**
@@ -102,6 +113,42 @@ public String runSql(String sql, Object... args)
}
}
+ /**
+ * Runs the given SQL query for a datasource and verifies the result.
+ *
+ * @param query Must contain a {@code %s} placeholder for the datasource.
+ * @param dataSource Datasource for which the query should be run.
+ * @param expectedResultCsv Expected result as a CSV String.
+ */
+ public void verifySqlQuery(String query, String dataSource, String expectedResultCsv)
+ {
+ Assertions.assertEquals(
+ expectedResultCsv,
+ cluster.runSql(query, dataSource),
+ StringUtils.format("Query[%s] failed", query)
+ );
+ }
+
+ /**
+ * Creates a Task using the given builder and runs it.
+ *
+ * @return ID of the task.
+ */
+ public String runTask(
+ TaskBuilder taskBuilder,
+ String dataSource,
+ EmbeddedOverlord overlord,
+ EmbeddedCoordinator coordinator
+ )
+ {
+ final String taskId = EmbeddedClusterApis.newTaskId(dataSource);
+ onLeaderOverlord(o -> o.runTask(taskId, taskBuilder.build(dataSource, taskId)));
+ waitForTaskToSucceed(taskId, overlord);
+ waitForAllSegmentsToBeAvailable(dataSource, coordinator);
+
+ return taskId;
+ }
+
/**
* Waits for the given task to finish successfully. If the given
* {@link EmbeddedOverlord} is not the leader, this method can only return by
@@ -126,6 +173,83 @@ public void waitForTaskToFinish(String taskId, EmbeddedOverlord overlord)
);
}
+ /**
+ * Retrieves all used segments from the metadata store (or cache if applicable).
+ */
+ public Set getVisibleUsedSegments(String dataSource, EmbeddedOverlord overlord)
+ {
+ return overlord
+ .bindings()
+ .segmentsMetadataStorage()
+ .retrieveAllUsedSegments(dataSource, Segments.ONLY_VISIBLE);
+ }
+
+ /**
+ * Returns intervals of all visible used segments sorted using the
+ * {@link Comparators#intervalsByStartThenEnd()}.
+ */
+ public List getSortedSegmentIntervals(String dataSource, EmbeddedOverlord overlord)
+ {
+ final Comparator comparator = Comparators.intervalsByStartThenEnd().reversed();
+ final TreeSet sortedIntervals = new TreeSet<>(comparator);
+
+ final Set allUsedSegments = getVisibleUsedSegments(dataSource, overlord);
+ for (DataSegment segment : allUsedSegments) {
+ sortedIntervals.add(segment.getInterval());
+ }
+
+ return new ArrayList<>(sortedIntervals);
+ }
+
+ /**
+ * Verifies that the number of visible used segments is the same as expected.
+ */
+ public void verifyNumVisibleSegmentsIs(int numExpectedSegments, String dataSource, EmbeddedOverlord overlord)
+ {
+ int segmentCount = cluster.callApi().getVisibleUsedSegments(dataSource, overlord).size();
+ Assertions.assertEquals(
+ numExpectedSegments,
+ segmentCount,
+ "Segment count mismatch"
+ );
+ Assertions.assertEquals(
+ String.valueOf(segmentCount),
+ cluster.runSql(
+ "SELECT COUNT(*) FROM sys.segments WHERE datasource='%s'"
+ + " AND is_overshadowed = 0 AND is_available = 1",
+ dataSource
+ ),
+ "Segment count mismatch in sys.segments table"
+ );
+ }
+
+ /**
+ * Waits for all used segments (including overshadowed) of the given datasource
+ * to be loaded on historicals.
+ */
+ public void waitForAllSegmentsToBeAvailable(String dataSource, EmbeddedCoordinator coordinator)
+ {
+ final int numSegments = coordinator
+ .bindings()
+ .segmentsMetadataStorage()
+ .retrieveAllUsedSegments(dataSource, Segments.INCLUDING_OVERSHADOWED)
+ .size();
+ coordinator.latchableEmitter().waitForEventAggregate(
+ event -> event.hasMetricName("segment/loadQueue/success")
+ .hasDimension(DruidMetrics.DATASOURCE, dataSource),
+ agg -> agg.hasSumAtLeast(numSegments)
+ );
+ }
+
+ /**
+ * Returns a {@link Closeable} that deletes all the data for the given datasource
+ * on {@link Closeable#close()}.
+ */
+ public Closeable createUnloader(String dataSource)
+ {
+ return () -> onLeaderOverlord(o -> o.markSegmentsAsUnused(dataSource));
+ }
+
/**
* Fetches the status of the given task from the cluster and verifies that it
* matches the expected status.
@@ -165,6 +289,8 @@ public SupervisorStatus getSupervisorStatus(String supervisorId)
throw new ISE("Could not find supervisor[%s]", supervisorId);
}
+ // STATIC UTILITY METHODS
+
/**
* Creates a random datasource name prefixed with {@link TestDataSource#WIKI}.
*/
@@ -174,15 +300,11 @@ public static String createTestDatasourceName()
}
/**
- * Deserializes the given String payload into a Map-based object that may be
- * submitted directly to the Overlord using
- * {@code cluster.callApi().onLeaderOverlord(o -> o.runTask(...))}.
+ * Creates a random task ID prefixed with the {@code dataSource}.
*/
- public static Object createTaskFromPayload(String taskId, String payload)
+ public static String newTaskId(String dataSource)
{
- final Map task = deserializeJsonToMap(payload);
- task.put("id", taskId);
- return task;
+ return dataSource + "_" + IdUtils.getRandomId();
}
/**
@@ -200,8 +322,35 @@ public static Map deserializeJsonToMap(String payload)
}
}
+ /**
+ * Creates a list of intervals that align with the given target granularity
+ * and overlap the original list of given intervals. If the original list is
+ * sorted, the returned list would be sorted too.
+ */
+ public static List createAlignedIntervals(
+ List original,
+ Granularity targetGranularity
+ )
+ {
+ final List alignedIntervals = new ArrayList<>();
+ for (Interval interval : original) {
+ for (Interval alignedInterval :
+ targetGranularity.getIterable(new Interval(interval, ISOChronology.getInstanceUTC()))) {
+ alignedIntervals.add(alignedInterval);
+ }
+ }
+
+ return alignedIntervals;
+ }
+
private static T getResult(ListenableFuture future)
{
return FutureUtils.getUnchecked(future, true);
}
+
+ @FunctionalInterface
+ public interface TaskBuilder
+ {
+ Object build(String dataSource, String taskId);
+ }
}
diff --git a/services/src/test/java/org/apache/druid/testing/embedded/derby/InMemoryDerbyModule.java b/services/src/test/java/org/apache/druid/testing/embedded/derby/InMemoryDerbyModule.java
index 1ad787706076..addc2c1919aa 100644
--- a/services/src/test/java/org/apache/druid/testing/embedded/derby/InMemoryDerbyModule.java
+++ b/services/src/test/java/org/apache/druid/testing/embedded/derby/InMemoryDerbyModule.java
@@ -74,7 +74,7 @@ public void configure(Binder binder)
MetadataStorageConnectorConfig.create(connectURI, null, null, null),
MetadataStorageTablesConfig.fromBase(properties.getProperty("druid.metadata.storage.tables.base")),
connectURI,
- CentralizedDatasourceSchemaConfig.create()
+ createSchemaConfig()
);
PolyBind.optionBinder(binder, Key.get(MetadataStorageProvider.class))
@@ -95,4 +95,12 @@ public void configure(Binder binder)
.to(DerbyMetadataStorageActionHandlerFactory.class)
.in(LazySingleton.class);
}
+
+ private CentralizedDatasourceSchemaConfig createSchemaConfig()
+ {
+ final Boolean isEnabled = Boolean.parseBoolean(
+ properties.getProperty("druid.centralizedDatasourceSchema.enabled")
+ );
+ return new CentralizedDatasourceSchemaConfig(isEnabled, null, null, null);
+ }
}
diff --git a/services/src/test/java/org/apache/druid/testing/embedded/indexing/Resources.java b/services/src/test/java/org/apache/druid/testing/embedded/indexing/Resources.java
index 3e4a7e1f6236..7af3d3a7c7a2 100644
--- a/services/src/test/java/org/apache/druid/testing/embedded/indexing/Resources.java
+++ b/services/src/test/java/org/apache/druid/testing/embedded/indexing/Resources.java
@@ -24,63 +24,74 @@
*/
public class Resources
{
- public static final String CSV_DATA_10_DAYS =
- "2025-06-01T00:00:00.000Z,shirt,105"
- + "\n2025-06-02T00:00:00.000Z,trousers,210"
- + "\n2025-06-03T00:00:00.000Z,jeans,150"
- + "\n2025-06-04T00:00:00.000Z,t-shirt,53"
- + "\n2025-06-05T00:00:00.000Z,microwave,1099"
- + "\n2025-06-06T00:00:00.000Z,spoon,11"
- + "\n2025-06-07T00:00:00.000Z,television,1100"
- + "\n2025-06-08T00:00:00.000Z,plant pots,75"
- + "\n2025-06-09T00:00:00.000Z,shirt,99"
- + "\n2025-06-10T00:00:00.000Z,toys,101";
+ public static class InlineData
+ {
+ /**
+ * 10 rows (1 row per day) of inline CSV data with 3 columns (time, item, value).
+ */
+ public static final String CSV_10_DAYS =
+ "2025-06-01T00:00:00.000Z,shirt,105"
+ + "\n2025-06-02T00:00:00.000Z,trousers,210"
+ + "\n2025-06-03T00:00:00.000Z,jeans,150"
+ + "\n2025-06-04T00:00:00.000Z,t-shirt,53"
+ + "\n2025-06-05T00:00:00.000Z,microwave,1099"
+ + "\n2025-06-06T00:00:00.000Z,spoon,11"
+ + "\n2025-06-07T00:00:00.000Z,television,1100"
+ + "\n2025-06-08T00:00:00.000Z,plant pots,75"
+ + "\n2025-06-09T00:00:00.000Z,shirt,99"
+ + "\n2025-06-10T00:00:00.000Z,toys,101";
+
+ public static final String JSON_2_ROWS =
+ "{\"isRobot\":true,\"language\":\"en\",\"timestamp\":\"2013-08-31T00:00:11.080Z\","
+ + "\"flags\":\"NB\",\"isUnpatrolled\":false,\"page\":\"Salo Toraut\","
+ + "\"diffUrl\":\"https://sv.wikipedia.org/w/index.php?oldid=36099284&rcid=89369918\","
+ + "\"added\":31,\"comment\":\"Botskapande Indonesien omdirigering\","
+ + "\"commentLength\":35,\"isNew\":true,\"isMinor\":false,\"delta\":31,"
+ + "\"isAnonymous\":false,\"user\":\"maytas1\",\"deltaBucket\":0.0,\"deleted\":0,"
+ + "\"namespace\":\"Main\"}"
+ + "\n{\"isRobot\":true,\"language\":\"en\",\"timestamp\":\"2013-08-31T00:00:11.080Z\","
+ + "\"flags\":\"NB\",\"isUnpatrolled\":false,\"page\":\"Salo Toraut\","
+ + "\"diffUrl\":\"https://sv.wikipedia.org/w/index.php?oldid=36099284&rcid=89369918\","
+ + "\"added\":31,\"comment\":\"Botskapande Indonesien omdirigering\",\"commentLength\":35,"
+ + "\"isNew\":true,\"isMinor\":false,\"delta\":11,\"isAnonymous\":false,\"user\":\"maytas2\","
+ + "\"deltaBucket\":0.0,\"deleted\":0,\"namespace\":\"Main\"}\n";
+
+ public static final String JSON_1_ROW =
+ "{\"isRobot\":true,\"language\":\"en\",\"timestamp\":\"2013-08-31T00:00:11.080Z\","
+ + "\"flags\":\"NB\",\"isUnpatrolled\":false,\"page\":\"Salo Toraut\","
+ + "\"diffUrl\":\"https://sv.wikipedia.org/w/index.php?oldid=36099284&rcid=89369918\","
+ + "\"added\":31,\"comment\":\"Botskapande Indonesien omdirigering\","
+ + "\"commentLength\":35,\"isNew\":true,\"isMinor\":false,\"delta\":31,"
+ + "\"isAnonymous\":false,\"user\":\"maytas3\",\"deltaBucket\":0.0,\"deleted\":0,\"namespace\":\"Main\"}\n";
+ }
+
+ public static class DataFile
+ {
+ public static final String TINY_WIKI_1_JSON = "data/json/tiny_wiki_1.json";
+ public static final String TINY_WIKI_2_JSON = "data/json/tiny_wiki_2.json";
+ public static final String TINY_WIKI_3_JSON = "data/json/tiny_wiki_3.json";
+ }
/**
- * Full task payload for an "index" task. The payload has the following format
- * arguments:
- *
- * - Data which can be provided as a single CSV string. The data is expected
- * to have 3 columns: "time", "item" and "value". (e.g {@link #CSV_DATA_10_DAYS})
- * - Datasource name
- *
+ * Contains queries (that can be run with {@code cluster.runSql(...)}) and
+ * their corresponding results.
*/
- public static final String INDEX_TASK_PAYLOAD_WITH_INLINE_DATA
- = "{"
- + " \"type\": \"index\","
- + " \"spec\": {"
- + " \"ioConfig\": {"
- + " \"type\": \"index\","
- + " \"inputSource\": {"
- + " \"type\": \"inline\","
- + " \"data\": \"%s\""
- + " },\n"
- + " \"inputFormat\": {"
- + " \"type\": \"csv\","
- + " \"findColumnsFromHeader\": false,"
- + " \"columns\": [\"time\",\"item\",\"value\"]"
- + " }"
- + " },"
- + " \"tuningConfig\": {"
- + " \"type\": \"index_parallel\","
- + " \"partitionsSpec\": {"
- + " \"type\": \"dynamic\","
- + " \"maxRowsPerSegment\": 1000"
- + " }"
- + " },"
- + " \"dataSchema\": {"
- + " \"dataSource\": \"%s\","
- + " \"timestampSpec\": {"
- + " \"column\": \"time\","
- + " \"format\": \"iso\""
- + " },"
- + " \"dimensionsSpec\": {"
- + " \"dimensions\": []"
- + " },"
- + " \"granularitySpec\": {"
- + " \"segmentGranularity\": \"DAY\""
- + " }"
- + " }"
- + " }"
- + "}";
+ public static class Query
+ {
+ // Queries used with Task.BASIC_INDEX
+ public static final String SELECT_MIN_MAX_TIME = "SELECT MIN(__time), MAX(__time) FROM %s";
+ public static final String SELECT_APPROX_COUNT_DISTINCT =
+ "SELECT"
+ + " APPROX_COUNT_DISTINCT_DS_THETA(\"thetaSketch\"),"
+ + " APPROX_COUNT_DISTINCT_DS_HLL(\"HLLSketchBuild\")"
+ + " FROM %s";
+ public static final String SELECT_EARLIEST_LATEST_USER =
+ "SELECT EARLIEST(\"user\"), LATEST(\"user\") FROM %s WHERE __time < '2013-09-01'";
+ public static final String SELECT_COUNT_OF_CHINESE_PAGES =
+ "SELECT \"page\", COUNT(*) AS \"rows\", SUM(\"added\"), 10 * SUM(\"added\") AS added_times_ten"
+ + " FROM %s"
+ + " WHERE \"language\" = 'zh' AND __time < '2013-09-01'"
+ + " GROUP BY 1"
+ + " HAVING added_times_ten > 9000";
+ }
}