This is an automated email from the ASF dual-hosted git repository.
jihoonson pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-druid.git
The following commit(s) were added to refs/heads/master by this push:
new a81b1b8 index_parallel: support !appendToExisting with no explicit
intervals (#7046)
a81b1b8 is described below
commit a81b1b8c9c9ccf60ea28a2c2ffeedbc8dc19de5e
Author: David Glasser <[email protected]>
AuthorDate: Wed Feb 20 10:54:26 2019 -0800
index_parallel: support !appendToExisting with no explicit intervals (#7046)
* index_parallel: support !appendToExisting with no explicit intervals
This enables ParallelIndexSupervisorTask to dynamically request locks at
runtime
if it is run without explicit intervals in the granularity spec and with
appendToExisting set to false. Previously, it behaved as if
appendToExisting
was set to true, which was undocumented and inconsistent with IndexTask and
Hadoop indexing.
Also, when ParallelIndexSupervisorTask allocates segments in the explicit
interval case, fail if its locks on the interval have been revoked.
Also make a few other additions/clarifications to native ingestion docs.
Fixes #6989.
* Review feedback.
PR description on GitHub updated to match.
* Make native batch ingestion partitions start at 0
* Fix to previous commit
* Unit test. Verified to fail without the other commits on this branch.
* Another round of review
* Slightly scarier warning
---
docs/content/ingestion/ingestion-spec.md | 4 +-
docs/content/ingestion/native_tasks.md | 27 +++++++++
.../org/apache/druid/indexing/common/Counters.java | 4 +-
.../task/batch/parallel/ParallelIndexSubTask.java | 3 +-
.../parallel/ParallelIndexSupervisorTask.java | 70 +++++++++++++++++-----
.../indexing/common/task/IngestionTestBase.java | 5 ++
.../parallel/ParallelIndexSupervisorTaskTest.java | 28 ++++++++-
.../clients/CoordinatorResourceTestClient.java | 27 +++++++++
.../tests/indexer/AbstractITBatchIndexTest.java | 25 ++++++--
.../apache/druid/tests/indexer/ITIndexerTest.java | 3 +-
.../druid/tests/indexer/ITParallelIndexTest.java | 14 ++++-
.../wikipedia_parallel_reindex_queries.json | 18 ++++++
.../indexer/wikipedia_parallel_reindex_task.json | 65 ++++++++++++++++++++
13 files changed, 262 insertions(+), 31 deletions(-)
diff --git a/docs/content/ingestion/ingestion-spec.md
b/docs/content/ingestion/ingestion-spec.md
index 46e7334..b578b54 100644
--- a/docs/content/ingestion/ingestion-spec.md
+++ b/docs/content/ingestion/ingestion-spec.md
@@ -286,7 +286,7 @@ This spec is used to generated segments with uniform
intervals.
| segmentGranularity | string | The granularity to create time chunks at.
Multiple segments can be created per time chunk. For example, with 'DAY'
`segmentGranularity`, the events of the same day fall into the same time chunk
which can be optionally further partitioned into multiple segments based on
other configurations and input size. See
[Granularity](../querying/granularities.html) for supported granularities.| no
(default == 'DAY') |
| queryGranularity | string | The minimum granularity to be able to query
results at and the granularity of the data inside the segment. E.g. a value of
"minute" will mean that data is aggregated at minutely granularity. That is, if
there are collisions in the tuple (minute(timestamp), dimensions), then it will
aggregate values together using the aggregators instead of storing individual
rows. A granularity of 'NONE' means millisecond granularity. See
[Granularity](../querying/granularit [...]
| rollup | boolean | rollup or not | no (default == true) |
-| intervals | string | A list of intervals for the raw data being ingested.
Ignored for real-time ingestion. | no. If specified, batch ingestion tasks may
skip determining partitions phase which results in faster ingestion. |
+| intervals | JSON string array | A list of intervals for the raw data being
ingested. Ignored for real-time ingestion. | no. If specified, Hadoop and
native non-parallel batch ingestion tasks may skip determining partitions phase
which results in faster ingestion; native parallel ingestion tasks can request
all their locks up-front instead of one by one. Batch ingestion will thrown
away any data not in the specified intervals. |
### Arbitrary Granularity Spec
@@ -296,7 +296,7 @@ This spec is used to generate segments with arbitrary
intervals (it tries to cre
|-------|------|-------------|----------|
| queryGranularity | string | The minimum granularity to be able to query
results at and the granularity of the data inside the segment. E.g. a value of
"minute" will mean that data is aggregated at minutely granularity. That is, if
there are collisions in the tuple (minute(timestamp), dimensions), then it will
aggregate values together using the aggregators instead of storing individual
rows. A granularity of 'NONE' means millisecond granularity. See
[Granularity](../querying/granularit [...]
| rollup | boolean | rollup or not | no (default == true) |
-| intervals | string | A list of intervals for the raw data being ingested.
Ignored for real-time ingestion. | no. If specified, batch ingestion tasks may
skip determining partitions phase which results in faster ingestion. |
+| intervals | JSON string array | A list of intervals for the raw data being
ingested. Ignored for real-time ingestion. | no. If specified, Hadoop and
native non-parallel batch ingestion tasks may skip determining partitions phase
which results in faster ingestion; native parallel ingestion tasks can request
all their locks up-front instead of one by one. Batch ingestion will thrown
away any data not in the specified intervals. |
# Transform Spec
diff --git a/docs/content/ingestion/native_tasks.md
b/docs/content/ingestion/native_tasks.md
index 963adea..5d7ffa1 100644
--- a/docs/content/ingestion/native_tasks.md
+++ b/docs/content/ingestion/native_tasks.md
@@ -30,6 +30,9 @@ MiddleManager.
Please check [Hadoop-based Batch Ingestion VS Native Batch
Ingestion](./hadoop-vs-native-batch.html) for differences between native batch
ingestion and Hadoop-based ingestion.
+To run either kind of native batch indexing task, write an ingestion spec as
specified below. Then POST it to the
+[`/druid/indexer/v1/task` endpoint on the
Overlord](../operations/api-reference.html#tasks), or use the `post-index-task`
script included with Druid.
+
Parallel Index Task
--------------------------------
@@ -124,6 +127,11 @@ An example ingestion spec is:
}
```
+By default, batch ingestion replaces all data in any segment that it writes
to. If you'd like to add to the segment
+instead, set the appendToExisting flag in ioConfig. Note that it only replaces
data in segments where it actively adds
+data: if there are segments in your granularitySpec's intervals that have no
data written by this task, they will be
+left alone.
+
#### Task Properties
|property|description|required?|
@@ -139,6 +147,14 @@ This field is required.
See [Ingestion Spec DataSchema](../ingestion/ingestion-spec.html#dataschema)
+If you specify `intervals` explicitly in your dataSchema's granularitySpec,
batch ingestion will lock the full intervals
+specified when it starts up, and you will learn quickly if the specified
interval overlaps with locks held by other
+tasks (eg, Kafka ingestion). Otherwise, batch ingestion will lock each
interval as it is discovered, so you may only
+learn that the task overlaps with a higher-priority task later in ingestion.
If you specify `intervals` explicitly, any
+rows outside the specified intervals will be thrown away. We recommend setting
`intervals` explicitly if you know the
+time range of the data so that locking failure happens faster, and so that you
don't accidentally replace data outside
+that range if there's some stray data with unexpected timestamps.
+
#### IOConfig
|property|description|default|required?|
@@ -463,6 +479,11 @@ The Local Index Task is designed to be used for smaller
data sets. The task exec
}
```
+By default, batch ingestion replaces all data in any segment that it writes
to. If you'd like to add to the segment
+instead, set the appendToExisting flag in ioConfig. Note that it only replaces
data in segments where it actively adds
+data: if there are segments in your granularitySpec's intervals that have no
data written by this task, they will be
+left alone.
+
#### Task Properties
|property|description|required?|
@@ -478,6 +499,12 @@ This field is required.
See [Ingestion Spec DataSchema](../ingestion/ingestion-spec.html#dataschema)
+If you do not specify `intervals` explicitly in your dataSchema's
granularitySpec, the Local Index Task will do an extra
+pass over the data to determine the range to lock when it starts up. If you
specify `intervals` explicitly, any rows
+outside the specified intervals will be thrown away. We recommend setting
`intervals` explicitly if you know the time
+range of the data because it allows the task to skip the extra pass, and so
that you don't accidentally replace data outside
+that range if there's some stray data with unexpected timestamps.
+
#### IOConfig
|property|description|default|required?|
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/Counters.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/Counters.java
index e463890..a57f419 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/Counters.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/Counters.java
@@ -25,7 +25,7 @@ import java.util.concurrent.atomic.AtomicLong;
public final class Counters
{
- public static <K> int incrementAndGetInt(ConcurrentHashMap<K, AtomicInteger>
counters, K key)
+ public static <K> int getAndIncrementInt(ConcurrentHashMap<K, AtomicInteger>
counters, K key)
{
// get() before computeIfAbsent() is an optimization to avoid locking in
computeIfAbsent() if not needed.
// See
https://github.com/apache/incubator-druid/pull/6898#discussion_r251384586.
@@ -33,7 +33,7 @@ public final class Counters
if (counter == null) {
counter = counters.computeIfAbsent(key, k -> new AtomicInteger());
}
- return counter.incrementAndGet();
+ return counter.getAndIncrement();
}
public static <K> long incrementAndGetLong(ConcurrentHashMap<K, AtomicLong>
counters, K key)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSubTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSubTask.java
index 18e87d7..8004243 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSubTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSubTask.java
@@ -235,9 +235,8 @@ public class ParallelIndexSubTask extends AbstractTask
)
{
final DataSchema dataSchema = ingestionSchema.getDataSchema();
- final boolean explicitIntervals =
dataSchema.getGranularitySpec().bucketIntervals().isPresent();
final ParallelIndexIOConfig ioConfig = ingestionSchema.getIOConfig();
- if (ioConfig.isAppendToExisting() || !explicitIntervals) {
+ if (ioConfig.isAppendToExisting()) {
return new ActionBasedSegmentAllocator(
toolbox.getTaskActionClient(),
dataSchema,
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
index 385797f..61f4517 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
@@ -33,8 +33,10 @@ import org.apache.druid.data.input.FirehoseFactory;
import org.apache.druid.indexer.TaskStatus;
import org.apache.druid.indexing.common.Counters;
import org.apache.druid.indexing.common.TaskLock;
+import org.apache.druid.indexing.common.TaskLockType;
import org.apache.druid.indexing.common.TaskToolbox;
import org.apache.druid.indexing.common.actions.LockListAction;
+import org.apache.druid.indexing.common.actions.LockTryAcquireAction;
import org.apache.druid.indexing.common.actions.TaskActionClient;
import org.apache.druid.indexing.common.stats.RowIngestionMetersFactory;
import org.apache.druid.indexing.common.task.AbstractTask;
@@ -360,43 +362,79 @@ public class ParallelIndexSupervisorTask extends
AbstractTask implements ChatHan
{
final String dataSource = getDataSource();
final GranularitySpec granularitySpec =
getIngestionSchema().getDataSchema().getGranularitySpec();
- final SortedSet<Interval> bucketIntervals = Preconditions.checkNotNull(
- granularitySpec.bucketIntervals().orNull(),
- "bucketIntervals"
- );
+ final Optional<SortedSet<Interval>> bucketIntervals =
granularitySpec.bucketIntervals();
+
// List locks whenever allocating a new segment because locks might be
revoked and no longer valid.
- final Map<Interval, String> versions = toolbox
+ final List<TaskLock> locks = toolbox
.getTaskActionClient()
- .submit(new LockListAction())
+ .submit(new LockListAction());
+ final TaskLock revokedLock =
locks.stream().filter(TaskLock::isRevoked).findAny().orElse(null);
+ if (revokedLock != null) {
+ throw new ISE("Lock revoked: [%s]", revokedLock);
+ }
+ final Map<Interval, String> versions = locks
.stream()
.collect(Collectors.toMap(TaskLock::getInterval,
TaskLock::getVersion));
- final Optional<Interval> maybeInterval =
granularitySpec.bucketInterval(timestamp);
- if (!maybeInterval.isPresent()) {
- throw new IAE("Could not find interval for timestamp [%s]", timestamp);
- }
+ Interval interval;
+ String version;
+ boolean justLockedInterval = false;
+ if (bucketIntervals.isPresent()) {
+ // If the granularity spec has explicit intervals, we just need to find
the interval (of the segment
+ // granularity); we already tried to lock it at task startup.
+ final Optional<Interval> maybeInterval =
granularitySpec.bucketInterval(timestamp);
+ if (!maybeInterval.isPresent()) {
+ throw new IAE("Could not find interval for timestamp [%s]", timestamp);
+ }
+
+ interval = maybeInterval.get();
+ if (!bucketIntervals.get().contains(interval)) {
+ throw new ISE("Unspecified interval[%s] in granularitySpec[%s]",
interval, granularitySpec);
+ }
- final Interval interval = maybeInterval.get();
- if (!bucketIntervals.contains(interval)) {
- throw new ISE("Unspecified interval[%s] in granularitySpec[%s]",
interval, granularitySpec);
+ version = findVersion(versions, interval);
+ if (version == null) {
+ throw new ISE("Cannot find a version for interval[%s]", interval);
+ }
+ } else {
+ // We don't have explicit intervals. We can use the segment granularity
to figure out what
+ // interval we need, but we might not have already locked it.
+ interval = granularitySpec.getSegmentGranularity().bucket(timestamp);
+ version = findVersion(versions, interval);
+ if (version == null) {
+ // We don't have a lock for this interval, so we should lock it now.
+ final TaskLock lock = Preconditions.checkNotNull(
+ toolbox.getTaskActionClient().submit(new
LockTryAcquireAction(TaskLockType.EXCLUSIVE, interval)),
+ "Cannot acquire a lock for interval[%s]", interval
+ );
+ version = lock.getVersion();
+ justLockedInterval = true;
+ }
}
- final int partitionNum =
Counters.incrementAndGetInt(partitionNumCountersPerInterval, interval);
+ final int partitionNum =
Counters.getAndIncrementInt(partitionNumCountersPerInterval, interval);
+ if (justLockedInterval && partitionNum != 0) {
+ throw new ISE(
+ "Expected partitionNum to be 0 for interval [%s] right after
locking, but got [%s]",
+ interval, partitionNum
+ );
+ }
return new SegmentIdWithShardSpec(
dataSource,
interval,
- findVersion(versions, interval),
+ version,
new NumberedShardSpec(partitionNum, 0)
);
}
+ @Nullable
private static String findVersion(Map<Interval, String> versions, Interval
interval)
{
return versions.entrySet().stream()
.filter(entry -> entry.getKey().contains(interval))
.map(Entry::getValue)
.findFirst()
- .orElseThrow(() -> new ISE("Cannot find a version for
interval[%s]", interval));
+ .orElse(null);
}
/**
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java
index 8406adb..c70a71a 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java
@@ -113,6 +113,11 @@ public abstract class IngestionTestBase
return lockbox;
}
+ public IndexerSQLMetadataStorageCoordinator getStorageCoordinator()
+ {
+ return storageCoordinator;
+ }
+
public TaskActionToolbox createTaskActionToolbox()
{
storageCoordinator.start();
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskTest.java
index 1d25f73..241e9f5 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskTest.java
@@ -36,6 +36,7 @@ import
org.apache.druid.query.aggregation.LongSumAggregatorFactory;
import org.apache.druid.segment.indexing.DataSchema;
import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec;
import org.apache.druid.segment.realtime.firehose.LocalFirehoseFactory;
+import org.apache.druid.timeline.DataSegment;
import org.joda.time.Interval;
import org.junit.After;
import org.junit.Assert;
@@ -51,6 +52,7 @@ import java.nio.file.Files;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
public class ParallelIndexSupervisorTaskTest extends
AbstractParallelIndexSupervisorTaskTest
@@ -126,8 +128,7 @@ public class ParallelIndexSupervisorTaskTest extends
AbstractParallelIndexSuperv
}
}
- @Test
- public void testWithoutInterval() throws Exception
+ private void runTestWithoutIntervalTask() throws Exception
{
final ParallelIndexSupervisorTask task = newTask(
null,
@@ -142,6 +143,29 @@ public class ParallelIndexSupervisorTaskTest extends
AbstractParallelIndexSuperv
prepareTaskForLocking(task);
Assert.assertTrue(task.isReady(actionClient));
Assert.assertEquals(TaskState.SUCCESS, task.run(toolbox).getStatusCode());
+ shutdownTask(task);
+ }
+
+ @Test
+ public void testWithoutInterval() throws Exception
+ {
+ // Ingest all data.
+ runTestWithoutIntervalTask();
+
+ // Read the segments for one day.
+ final Interval interval = Intervals.of("2017-12-24/P1D");
+ final List<DataSegment> oldSegments =
+ getStorageCoordinator().getUsedSegmentsForInterval("dataSource",
interval);
+ Assert.assertEquals(1, oldSegments.size());
+
+ // Reingest the same data. Each segment should get replaced by a segment
with a newer version.
+ runTestWithoutIntervalTask();
+
+ // Verify that the segment has been replaced.
+ final List<DataSegment> newSegments =
+ getStorageCoordinator().getUsedSegmentsForInterval("dataSource",
interval);
+ Assert.assertEquals(1, newSegments.size());
+
Assert.assertTrue(oldSegments.get(0).getVersion().compareTo(newSegments.get(0).getVersion())
< 0);
}
@Test()
diff --git
a/integration-tests/src/main/java/org/apache/druid/testing/clients/CoordinatorResourceTestClient.java
b/integration-tests/src/main/java/org/apache/druid/testing/clients/CoordinatorResourceTestClient.java
index df6ad20..04147b0 100644
---
a/integration-tests/src/main/java/org/apache/druid/testing/clients/CoordinatorResourceTestClient.java
+++
b/integration-tests/src/main/java/org/apache/druid/testing/clients/CoordinatorResourceTestClient.java
@@ -32,6 +32,7 @@ import
org.apache.druid.java.util.http.client.response.StatusResponseHandler;
import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
import org.apache.druid.testing.IntegrationTestingConfig;
import org.apache.druid.testing.guice.TestClient;
+import org.apache.druid.timeline.DataSegment;
import org.jboss.netty.handler.codec.http.HttpMethod;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import org.joda.time.Interval;
@@ -41,6 +42,8 @@ import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
public class CoordinatorResourceTestClient
{
@@ -80,6 +83,11 @@ public class CoordinatorResourceTestClient
return StringUtils.format("%sdatasources/%s/intervals",
getCoordinatorURL(), StringUtils.urlEncode(dataSource));
}
+ private String getFullSegmentsURL(String dataSource)
+ {
+ return StringUtils.format("%sdatasources/%s/segments?full",
getCoordinatorURL(), StringUtils.urlEncode(dataSource));
+ }
+
private String getLoadStatusURL()
{
return StringUtils.format("%s%s", getCoordinatorURL(), "loadstatus");
@@ -123,6 +131,25 @@ public class CoordinatorResourceTestClient
return segments;
}
+ // return a set of the segment versions for the specified datasource
+ public Set<String> getSegmentVersions(final String dataSource)
+ {
+ ArrayList<DataSegment> segments;
+ try {
+ StatusResponseHolder response = makeRequest(HttpMethod.GET,
getFullSegmentsURL(dataSource));
+
+ segments = jsonMapper.readValue(
+ response.getContent(), new TypeReference<List<DataSegment>>()
+ {
+ }
+ );
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return segments.stream().map(s ->
s.getVersion()).collect(Collectors.toSet());
+ }
+
private Map<String, Integer> getLoadStatus()
{
Map<String, Integer> status;
diff --git
a/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITBatchIndexTest.java
b/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITBatchIndexTest.java
index 3555ef8..2034b60 100644
---
a/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITBatchIndexTest.java
+++
b/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITBatchIndexTest.java
@@ -33,6 +33,7 @@ import org.junit.Assert;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;
+import java.util.Set;
public class AbstractITBatchIndexTest extends AbstractIndexerTest
{
@@ -49,7 +50,8 @@ public class AbstractITBatchIndexTest extends
AbstractIndexerTest
void doIndexTestTest(
String dataSource,
String indexTaskFilePath,
- String queryFilePath
+ String queryFilePath,
+ boolean waitForNewVersion
) throws IOException
{
final String fullDatasourceName = dataSource +
config.getExtraDatasourceNameSuffix();
@@ -59,7 +61,7 @@ public class AbstractITBatchIndexTest extends
AbstractIndexerTest
fullDatasourceName
);
- submitTaskAndWait(taskSpec, fullDatasourceName);
+ submitTaskAndWait(taskSpec, fullDatasourceName, waitForNewVersion);
try {
String queryResponseTemplate;
@@ -107,7 +109,7 @@ public class AbstractITBatchIndexTest extends
AbstractIndexerTest
fullReindexDatasourceName
);
- submitTaskAndWait(taskSpec, fullReindexDatasourceName);
+ submitTaskAndWait(taskSpec, fullReindexDatasourceName, false);
try {
String queryResponseTemplate;
try {
@@ -144,7 +146,7 @@ public class AbstractITBatchIndexTest extends
AbstractIndexerTest
String queryFilePath
)
{
- submitTaskAndWait(indexTaskFilePath, dataSource);
+ submitTaskAndWait(indexTaskFilePath, dataSource, false);
try {
sqlQueryHelper.testQueriesFromFile(queryFilePath, 2);
}
@@ -154,12 +156,25 @@ public class AbstractITBatchIndexTest extends
AbstractIndexerTest
}
}
- private void submitTaskAndWait(String taskSpec, String dataSourceName)
+ private void submitTaskAndWait(String taskSpec, String dataSourceName,
boolean waitForNewVersion)
{
+ final Set<String> oldVersions = waitForNewVersion ?
coordinator.getSegmentVersions(dataSourceName) : null;
+
final String taskID = indexer.submitTask(taskSpec);
LOG.info("TaskID for loading index task %s", taskID);
indexer.waitUntilTaskCompletes(taskID);
+ // ITParallelIndexTest does a second round of ingestion to replace
segements in an existing
+ // data source. For that second round we need to make sure the coordinator
actually learned
+ // about the new segments befor waiting for it to report that all segments
are loaded; otherwise
+ // this method could return too early because the coordinator is merely
reporting that all the
+ // original segments have loaded.
+ if (waitForNewVersion) {
+ RetryUtil.retryUntilTrue(
+ () ->
!oldVersions.containsAll(coordinator.getSegmentVersions(dataSourceName)), "See
a new version"
+ );
+ }
+
RetryUtil.retryUntilTrue(
() -> coordinator.areSegmentsLoaded(dataSourceName), "Segment Load"
);
diff --git
a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITIndexerTest.java
b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITIndexerTest.java
index 8412e49..245c3dd 100644
---
a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITIndexerTest.java
+++
b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITIndexerTest.java
@@ -45,7 +45,8 @@ public class ITIndexerTest extends AbstractITBatchIndexTest
doIndexTestTest(
INDEX_DATASOURCE,
INDEX_TASK,
- INDEX_QUERIES_RESOURCE
+ INDEX_QUERIES_RESOURCE,
+ false
);
doReindexTest(
INDEX_DATASOURCE,
diff --git
a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITParallelIndexTest.java
b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITParallelIndexTest.java
index 80ca6e1..e457a1c 100644
---
a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITParallelIndexTest.java
+++
b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITParallelIndexTest.java
@@ -30,6 +30,8 @@ public class ITParallelIndexTest extends
AbstractITBatchIndexTest
{
private static String INDEX_TASK =
"/indexer/wikipedia_parallel_index_task.json";
private static String INDEX_QUERIES_RESOURCE =
"/indexer/wikipedia_parallel_index_queries.json";
+ private static String REINDEX_TASK =
"/indexer/wikipedia_parallel_reindex_task.json";
+ private static String REINDEX_QUERIES_RESOURCE =
"/indexer/wikipedia_parallel_reindex_queries.json";
private static String INDEX_DATASOURCE = "wikipedia_parallel_index_test";
@Test
@@ -39,7 +41,17 @@ public class ITParallelIndexTest extends
AbstractITBatchIndexTest
doIndexTestTest(
INDEX_DATASOURCE,
INDEX_TASK,
- INDEX_QUERIES_RESOURCE
+ INDEX_QUERIES_RESOURCE,
+ 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.
+ doIndexTestTest(
+ INDEX_DATASOURCE,
+ REINDEX_TASK,
+ REINDEX_QUERIES_RESOURCE,
+ true
);
}
}
diff --git
a/integration-tests/src/test/resources/indexer/wikipedia_parallel_reindex_queries.json
b/integration-tests/src/test/resources/indexer/wikipedia_parallel_reindex_queries.json
new file mode 100644
index 0000000..bbbeca9
--- /dev/null
+++
b/integration-tests/src/test/resources/indexer/wikipedia_parallel_reindex_queries.json
@@ -0,0 +1,18 @@
+[
+ {
+ "description": "timeseries, 1 agg, all should only show data2",
+ "query":{
+ "queryType" : "timeBoundary",
+ "dataSource": "%%DATASOURCE%%"
+ },
+ "expectedResults":[
+ {
+ "timestamp" : "2013-08-31T11:58:39.000Z",
+ "result" : {
+ "minTime" : "2013-08-31T11:58:39.000Z",
+ "maxTime" : "2013-09-01T01:02:33.000Z"
+ }
+ }
+ ]
+ }
+]
\ No newline at end of file
diff --git
a/integration-tests/src/test/resources/indexer/wikipedia_parallel_reindex_task.json
b/integration-tests/src/test/resources/indexer/wikipedia_parallel_reindex_task.json
new file mode 100644
index 0000000..c06890b
--- /dev/null
+++
b/integration-tests/src/test/resources/indexer/wikipedia_parallel_reindex_task.json
@@ -0,0 +1,65 @@
+{
+ "type": "index_parallel",
+ "spec": {
+ "dataSchema": {
+ "dataSource": "%%DATASOURCE%%",
+ "metricsSpec": [
+ {
+ "type": "count",
+ "name": "count"
+ },
+ {
+ "type": "doubleSum",
+ "name": "added",
+ "fieldName": "added"
+ },
+ {
+ "type": "doubleSum",
+ "name": "deleted",
+ "fieldName": "deleted"
+ },
+ {
+ "type": "doubleSum",
+ "name": "delta",
+ "fieldName": "delta"
+ }
+ ],
+ "granularitySpec": {
+ "segmentGranularity": "DAY",
+ "queryGranularity": "second"
+ },
+ "parser": {
+ "parseSpec": {
+ "format" : "json",
+ "timestampSpec": {
+ "column": "timestamp"
+ },
+ "dimensionsSpec": {
+ "dimensions": [
+ "page",
+ {"type": "string", "name": "language",
"createBitmapIndex": false},
+ "user",
+ "unpatrolled",
+ "newPage",
+ "robot",
+ "anonymous",
+ "namespace",
+ "continent",
+ "country",
+ "region",
+ "city"
+ ]
+ }
+ }
+ }
+ },
+ "ioConfig": {
+ "type": "index_parallel",
+ "firehose": {
+ "type": "local",
+ "baseDir": "/resources/data/batch_index",
+ "filter": "wikipedia_index_data2*"
+ }
+ }
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
