A Replacing task must read segments created before it acquired its lock #15085

AmatyaAvadhanula · 2023-10-04T10:40:52Z

Fixes duplication of data with concurrent append and replace jobs.

Description

Consider the following events:

-Segments S1, S2, S3 exist
-Compact acquires replace lock
-Append publishes a segment S4 which needs to be carried forward
-Compact task processes S1-S4 to create new segments
-Compact task publishes new segments and carries S4 forward

This can lead to the data in S4 being duplicated

Approach:

This PR introduces a new task action to fetch only those segments that were present before the REPLACE lock was acquired.

RetrieveSegmentsToReplaceAction:

Implement the method getSupervisorTaskId at the level of AbstractBatchSubtask.
Fetch all replace locks for datasource using getAllReplaceLocksForDatasource, and filter those that have the same supervisorId.
Fetch segments and created_date for the interval of the action.
For each segment, find a replace lock in the filtered set whose interval contains its own interval and filter based on created_date and lock_version

This new action is utilized by the DruidInputSource.

Release note

This PR has:

...ice/src/main/java/org/apache/druid/indexing/common/actions/RetrieveLockedSegmentsAction.java

server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java

...ice/src/main/java/org/apache/druid/indexing/common/actions/RetrieveLockedSegmentsAction.java

server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java

...ice/src/main/java/org/apache/druid/indexing/common/actions/RetrieveLockedSegmentsAction.java

kfaraz · 2023-10-05T12:31:21Z

@AmatyaAvadhanula , should we use the new action only when using REPLACE lock type?

AmatyaAvadhanula · 2023-10-05T12:58:46Z

should we use the new action only when using REPLACE lock type?

I think it would be harder to figure out which action to use within the DruidInputSource.
The current action is identical to the previous action (for a single interval) in cases where the task itself doesn't hold a replace lock over the interval

kfaraz · 2023-10-06T03:24:50Z

I think it would be harder to figure out which action to use within the DruidInputSource.

I guess the only case we care about is CompactionTask, where we can just the TaskLockType while constructing the DruidInputSource. If this input source is passed along to the sub tasks, the DruidInputSource deserialized in the sub tasks will know the lock type as well.

The reason this would be preferable is the new action would just have to fetch the segments that the task is going to replace, i.e. segments belonging to the interval on which this task has a replace lock and which were created before the lock was acquired. It would avoid duplication and confusion with the existing action.

Other cases:

Re-indexing from datasource A to B: None of the above applies, so we would just continue using the old action.
Re-indexing into the same datasource: Does this get launched as a compact task or a regular index_parallel task? If it is the latter, then while doing REPLACE, the submitted spec would need to declare REPLACE both in task context and in the io config.

AmatyaAvadhanula · 2023-10-06T03:42:19Z

Re-indexing into the same datasource: Does this get launched as a compact task or a regular index_parallel task? If it is the latter, then while doing REPLACE, the submitted spec would need to declare REPLACE both in task context and in the io config.

It is the latter.

segments belonging to the interval on which this task has a replace lock and which were created before the lock was acquired. It would avoid duplication and confusion with the existing action.

I think that the task can fetch segments without having a lock as well (cardinality phase with hash partitioning for example)

kfaraz · 2023-10-06T04:48:33Z

I think that the task can fetch segments without having a lock as well (cardinality phase with hash partitioning for example)

Isn't the replace lock acquired by the index_parallel supervisor task even before it has started running, i.e. in the isReady call?

AmatyaAvadhanula · 2023-10-06T04:55:13Z

Isn't the replace lock acquired by the index_parallel supervisor task even before it has started running, i.e. in the isReady call?

No, it depends on ingestionSchema.getDataSchema().getGranularitySpec().inputIntervals() to lock the interval which is often unspecified. In such cases the lock is acquired after the cardinality phase reads the segments

kfaraz · 2023-10-08T02:32:36Z

@AmatyaAvadhanula , since the new action deals with replace locks, it must be fired using the SurrogateTaskActionClient.

AmatyaAvadhanula · 2023-10-08T03:19:05Z

since the new action deals with replace locks, it must be fired using the SurrogateTaskActionClient.

Thanks, @kfaraz. I'll make the change.

…laceCommit

server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java

…laceCommit

kfaraz

Overall approach looks fine, left some comments for improvements.

.../src/main/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsToReplaceAction.java

kfaraz · 2023-10-16T18:34:41Z

.../src/main/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsToReplaceAction.java

+        if (replaceLock.getInterval().contains(segment.getInterval())
+            && replaceLock.getVersion().compareTo(createdDate) < 0) {
+          // If a REPLACE lock covers a segment but has a version less than the segment's created date, remove it
+          allSegmentsToBeReplaced.remove(segment);


Rather than going through the collection of segments and created dates twice (once to add everything, then to remove the invalid ones), start with an empty set and add only the valid entries to it.

kfaraz · 2023-10-16T18:39:45Z

.../src/main/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsToReplaceAction.java

+
+    Set<DataSegment> allSegmentsToBeReplaced = new HashSet<>();
+    segmentsAndCreatedDates.forEach(segmentAndCreatedDate -> allSegmentsToBeReplaced.add(segmentAndCreatedDate.lhs));
+    for (Pair<DataSegment, String> segmentAndCreatedDate : segmentsAndCreatedDates) {


Given that a single interval may have several segments, see if this loop can be improved by using a map from Interval to version to Set<DataSegment>.

Done. Thanks!

kfaraz · 2023-10-16T18:40:35Z

.../src/main/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsToReplaceAction.java

+      final DataSegment segment = segmentAndCreatedDate.lhs;
+      final String createdDate = segmentAndCreatedDate.rhs;
+      for (ReplaceTaskLock replaceLock : replaceLocksForTask) {
+        if (replaceLock.getInterval().contains(segment.getInterval())


I suppose we don't really have much of a choice here other than doing this processing in-memory since these filters would be too complicated to push down.

kfaraz · 2023-10-16T18:41:38Z

indexing-service/src/main/java/org/apache/druid/indexing/common/actions/TaskAction.java

@@ -40,6 +40,7 @@
    @JsonSubTypes.Type(name = "segmentTransactionalReplace", value = SegmentTransactionalReplaceAction.class),
    // Type name doesn't correspond to the name of the class for backward compatibility.
    @JsonSubTypes.Type(name = "segmentListUsed", value = RetrieveUsedSegmentsAction.class),
+    @JsonSubTypes.Type(name = "segmentListLocked", value = RetrieveSegmentsToReplaceAction.class),


segmentListToReplace?

Can I use retrieveSegmentsToReplace since the other actions do not match the class names for backward compatibility?

kfaraz · 2023-10-16T18:44:55Z

...src/main/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractBatchSubtask.java

  }

  /**
   * Returns the ID of {@link SubTaskSpec} that is assigned to this subtask.
   * This ID is used to identify duplicate work of retry tasks for the same spec.
   */
  public abstract String getSubtaskSpecId();
+
+  /**
+   * @return the id of the subtask's supervisor


Suggested change

* @return the id of the subtask's supervisor

* @return Task ID of the {@code ParallelIndexSupervisorTask} which launched this sub-task.

kfaraz · 2023-10-16T18:50:50Z

server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java

+        "SELECT created_date, payload FROM %1$s WHERE dataSource = :dataSource AND used = true"
+    );
+
+    final boolean intervalStartIsEternityStart = Intervals.ETERNITY.getStart().equals(interval.getStart());


Why do we need to do this special eternity handling? Are other methods in this class doing this too?

This is important as such an interval may be passed to the method as an argument.
Other methods in this class do not implement it as they assume that no interval is passed when fetching segments for ETERNITY.

kfaraz · 2023-10-16T18:55:46Z

.../src/main/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsToReplaceAction.java

+        .filter(lock -> supervisorId.equals(lock.getSupervisorTaskId()))
+        .collect(Collectors.toSet());
+
+    Collection<Pair<DataSegment, String>> segmentsAndCreatedDates =


Return early if there are no REPLACE locks.

.../src/main/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsToReplaceAction.java

abhishekagarwal87 · 2023-10-18T08:22:17Z

.../src/main/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsToReplaceAction.java

+    // The DruidInputSource can be used to read from one datasource and write to another.
+    // In such a case, the action can simply fetch all visible segments for the datasource and interval


Suggested change

// The DruidInputSource can be used to read from one datasource and write to another.

// In such a case, the action can simply fetch all visible segments for the datasource and interval

// The DruidInputSource can be used to read from one datasource and write to another.

// In such a case, the race condition described in class-level docs cannot occur and the action can simply fetch all visible segments for the datasource and interval

server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java

abhishekagarwal87 · 2023-10-18T09:18:55Z

.../src/main/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsToReplaceAction.java

+      final Map<String, Set<DataSegment>> createdToSegmentsMap = entry.getValue();
+      for (Map.Entry<String, Set<DataSegment>> createdAndSegments : createdToSegmentsMap.entrySet()) {
+        if (lockVersion == null || lockVersion.compareTo(createdAndSegments.getKey()) > 0) {
+          allSegmentsToBeReplaced.addAll(createdAndSegments.getValue());


we should add a log line perhaps when a segment wasn't considered because it was created after replace lock was acquired. I am assuming it will not be frequent.

.../src/main/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsToReplaceAction.java

AmatyaAvadhanula · 2023-10-19T05:42:11Z

@kfaraz @abhishekagarwal87 Thank you for the review!

…ck (apache#15085) * Replacing tasks must read segments created before they acquired their locks

…ck (#15085) (#15208) * Replacing tasks must read segments created before they acquired their locks

…ck (apache#15085) * Replacing tasks must read segments created before they acquired their locks

Replacing tasks must read segments created before it acquired its lock

b158ce8

github-actions bot added the Area - Ingestion label Oct 4, 2023

AmatyaAvadhanula added 2 commits October 5, 2023 09:52

Fetch locked segments only if the task has the same datasource

5e3db67

Add test

71d7f7f

AmatyaAvadhanula marked this pull request as ready for review October 5, 2023 05:20

AmatyaAvadhanula requested a review from kfaraz October 5, 2023 05:20

kfaraz requested changes Oct 5, 2023

View reviewed changes

AmatyaAvadhanula added 6 commits October 8, 2023 10:43

Merge remote-tracking branch 'upstream/master' into fixTransactionRep…

e9a633f

…laceCommit

Fix conflicts

312229e

Add interval-wise logic

63009cb

Rename action and add datasource check

fdc294f

Add UT

d6c0fb4

Merge remote-tracking branch 'upstream/master' into fixTransactionRep…

29c9000

…laceCommit

AmatyaAvadhanula requested review from kfaraz and removed request for kfaraz October 14, 2023 12:54

AmatyaAvadhanula added 3 commits October 16, 2023 13:03

Fix coverage

15a0a5b

Fix style

85a85f5

Add eternity handling

b06217d

LakshSingla added this to the 28.0 milestone Oct 16, 2023

github-advanced-security bot found potential problems Oct 16, 2023

View reviewed changes

server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java Fixed Show fixed Hide fixed

server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java Fixed Show fixed Hide fixed

AmatyaAvadhanula added 2 commits October 16, 2023 17:36

Fix tests

65b59fa

Trigger Build

fc7bc39

Merge remote-tracking branch 'upstream/master' into fixTransactionRep…

66ab521

…laceCommit

kfaraz reviewed Oct 16, 2023

View reviewed changes

AmatyaAvadhanula added 2 commits October 17, 2023 09:23

Address review feedback

2cb68d0

Iterate over entries of map instead of keys

ccf2196

AmatyaAvadhanula requested a review from kfaraz October 17, 2023 10:11

abhishekagarwal87 reviewed Oct 18, 2023

View reviewed changes

AmatyaAvadhanula added 4 commits October 18, 2023 16:46

Address feedback

f264ef8

Re-use query builder

27b282d

Remove unused method

d98453c

Simplify calls

57b54e6

abhishekagarwal87 approved these changes Oct 19, 2023

View reviewed changes

AmatyaAvadhanula merged commit a8febd4 into apache:master Oct 19, 2023
81 checks passed

AmatyaAvadhanula added a commit to AmatyaAvadhanula/druid that referenced this pull request Oct 19, 2023

A Replacing task must read segments created before it acquired its lo…

e04413c

…ck (apache#15085) * Replacing tasks must read segments created before they acquired their locks

AmatyaAvadhanula mentioned this pull request Oct 19, 2023

Backport #15085 #15208

Merged

AmatyaAvadhanula added a commit that referenced this pull request Oct 19, 2023

A Replacing task must read segments created before it acquired its lo…

0c4c1cf

…ck (#15085) (#15208) * Replacing tasks must read segments created before they acquired their locks

kfaraz deleted the fixTransactionReplaceCommit branch November 7, 2023 04:15

CaseyPan pushed a commit to CaseyPan/druid that referenced this pull request Nov 17, 2023

A Replacing task must read segments created before it acquired its lo…

469f6a8

…ck (apache#15085) * Replacing tasks must read segments created before they acquired their locks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

A Replacing task must read segments created before it acquired its lock #15085

A Replacing task must read segments created before it acquired its lock #15085

AmatyaAvadhanula commented Oct 4, 2023 •

edited

Loading

kfaraz commented Oct 5, 2023

AmatyaAvadhanula commented Oct 5, 2023 •

edited

Loading

kfaraz commented Oct 6, 2023

AmatyaAvadhanula commented Oct 6, 2023

kfaraz commented Oct 6, 2023

AmatyaAvadhanula commented Oct 6, 2023

kfaraz commented Oct 8, 2023

AmatyaAvadhanula commented Oct 8, 2023

kfaraz left a comment

kfaraz Oct 16, 2023

AmatyaAvadhanula Oct 17, 2023

kfaraz Oct 16, 2023

AmatyaAvadhanula Oct 17, 2023

kfaraz Oct 16, 2023

kfaraz Oct 16, 2023

AmatyaAvadhanula Oct 17, 2023

kfaraz Oct 16, 2023

AmatyaAvadhanula Oct 17, 2023

kfaraz Oct 16, 2023

AmatyaAvadhanula Oct 17, 2023

kfaraz Oct 16, 2023

AmatyaAvadhanula Oct 17, 2023

abhishekagarwal87 Oct 18, 2023

AmatyaAvadhanula Oct 18, 2023

abhishekagarwal87 Oct 18, 2023

AmatyaAvadhanula Oct 18, 2023

AmatyaAvadhanula commented Oct 19, 2023

	* @return the id of the subtask's supervisor
	* @return Task ID of the {@code ParallelIndexSupervisorTask} which launched this sub-task.

		// The DruidInputSource can be used to read from one datasource and write to another.
		// In such a case, the action can simply fetch all visible segments for the datasource and interval

A Replacing task must read segments created before it acquired its lock #15085

A Replacing task must read segments created before it acquired its lock #15085

Conversation

AmatyaAvadhanula commented Oct 4, 2023 • edited Loading

Description

Approach:

Release note

kfaraz commented Oct 5, 2023

AmatyaAvadhanula commented Oct 5, 2023 • edited Loading

kfaraz commented Oct 6, 2023

AmatyaAvadhanula commented Oct 6, 2023

kfaraz commented Oct 6, 2023

AmatyaAvadhanula commented Oct 6, 2023

kfaraz commented Oct 8, 2023

AmatyaAvadhanula commented Oct 8, 2023

kfaraz left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AmatyaAvadhanula commented Oct 19, 2023

AmatyaAvadhanula commented Oct 4, 2023 •

edited

Loading

AmatyaAvadhanula commented Oct 5, 2023 •

edited

Loading