Filter pending segments upgraded with transactional replace #15169

AmatyaAvadhanula · 2023-10-16T14:37:43Z

The pending segment upgrade introduced as part of #15039 upgrades all pending segments for a given interval.
This PR aims to optimize it by only upgrading those pending segments for the interval which correspond to active tasks of the supervisor.
This can be done by fetching the baseSequenceNames of the active and pending completion task groups of the supervisor correspdonding to the datasource, and upgrading only those pending segments that have a sequence_name which has one of the preivous baseSequenceNames as the prefix.

This PR has:

kfaraz

Left some comments.

Also, I would prefer the name activeRealtimeSequencePrefixes as it avoids ambiguity.

kfaraz · 2023-10-16T18:15:14Z

...nce/src/main/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisor.java

+  @Override
+  public Set<String> getActiveBaseSequenceNames()
+  {
+    throw new UnsupportedOperationException("Get Active sequence names is not supported in MaterializedViewSupervisor");


Nit: Probably don't need an explicit message here.

Please fix this as the message is redundant.

Fixed. Missed this earlier

.../main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

kfaraz · 2023-10-16T18:20:58Z

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

@@ -345,10 +345,12 @@ SegmentPublishResult commitReplaceSegments(
   * </ul>
   *
   * @param replaceSegments Segments being committed by a REPLACE task
+   * @param activeBaseSequenceNames of base sequence names of active / pending completion task groups of the supervisor


Suggested change

* @param activeBaseSequenceNames of base sequence names of active / pending completion task groups of the supervisor

* @param activeBaseSequenceNames Set of base sequence names of active and pending completion task groups of the supervisor (if any) for this datasource

kfaraz · 2023-10-16T18:23:18Z

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

@@ -660,6 +662,18 @@ private Map<SegmentIdWithShardSpec, SegmentIdWithShardSpec> upgradePendingSegmen
          : overlappingPendingSegments.entrySet()) {
        final SegmentIdWithShardSpec pendingSegmentId = overlappingPendingSegment.getKey();
        final String pendingSegmentSequence = overlappingPendingSegment.getValue();
+
+        boolean considerSequence = false;


Based on the offline discussion, I think we had decided to include the sequence name in the SELECT query itself, rather than filtering in-memory here.

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

+                .bind("sequence_name", sequenceName)
+                .bind("sequence_prev_id", pendingSegment.toString())
+                .bind("sequence_name_prev_id_sha1", BaseEncoding.base16().encode(
+                    Hashing.sha1()


kfaraz

New approach looks better, just need some code cleanup for approval.

kfaraz · 2023-10-20T05:43:29Z

...test/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisorSpecTest.java

@@ -217,6 +217,13 @@ public void testMaterializedViewSupervisorSpecCreated()
        Assert.assertTrue(e instanceof UnsupportedOperationException);
      }

+      try {


Use Assert.assertThrows instead and update the existing assertions in this method too.

kfaraz · 2023-10-23T05:54:22Z

...nce/src/main/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisor.java

+  @Override
+  public Set<String> getActiveBaseSequenceNames()
+  {
+    throw new UnsupportedOperationException("Get Active sequence names is not supported in MaterializedViewSupervisor");


Please fix this as the message is redundant.

kfaraz · 2023-10-23T06:02:39Z

...g-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java

@@ -110,6 +111,14 @@ public Optional<String> getActiveSupervisorIdForDatasourceWithAppendLock(String
    return Optional.absent();
  }

+  public Set<String> getActiveRealtimeSequencePrefixes(String activeSupervisorId)
+  {
+    if (!supervisors.containsKey(activeSupervisorId)) {


Please invert the if condition:

if contains: return value else return empty

kfaraz · 2023-10-23T06:04:34Z

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

@@ -347,10 +347,13 @@ SegmentPublishResult commitReplaceSegments(
   * </ul>
   *
   * @param replaceSegments Segments being committed by a REPLACE task
+   * @param activeRealtimeSequencePrefixes Set of base sequence names of active and pending completion task groups


Please update the javadoc to say prefixes instead of sequence names.

.../main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

kfaraz · 2023-10-23T06:08:06Z

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

   */
+  @VisibleForTesting


Avoid using this. It is okay to just test the method which invokes this internally.

The method invoking this internally would ignore segments that are not needed eventually, and it would be hard to test this optimization.

Sorry, It seems like it filters segments that have already been upgraded, and not all the unnecessary segments. I can make this change

kfaraz · 2023-10-23T06:09:15Z

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

+    }
+
+    final List<String> sequenceNamePrefixes = new ArrayList<>(sequenceNamePrefixFilter);
+    StringBuilder sql = new StringBuilder(


Use .append instead of + as it is already a StringBuilder.

would be much easier to read and understand the sql if written using a format string.

SELECT sequence, payload FROM %s WHERE dataSource=:dataSource AND start < :end AND end < :start AND (:sequenceFilterCondition)

Then build a separate string for sequenceFilterCondition which would basically be a bunch of filters tied by ORs.

Done. Thanks!

kfaraz · 2023-10-23T06:16:47Z

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

+    for (int i = 1; i < sequenceNamePrefixes.size(); i++) {
+      sql.append("(sequence_name LIKE ")
+         .append(StringUtils.format(":prefix%d", i))
+         .append(")")
+         .append(" OR ");
+    }
+
+    sql.append("(sequence_name LIKE ")
+       .append(StringUtils.format(":prefix%d", sequenceNamePrefixes.size()))
+       .append(")");


Cleaner to just do this for i = 0 to sequenceNamePrefixes.size():

Suggested change

for (int i = 1; i < sequenceNamePrefixes.size(); i++) {

sql.append("(sequence_name LIKE ")

.append(StringUtils.format(":prefix%d", i))

.append(")")

.append(" OR ");

}

sql.append("(sequence_name LIKE ")

.append(StringUtils.format(":prefix%d", sequenceNamePrefixes.size()))

.append(")");

sequenceFilterCondition.append(

StringUtils.format("(sequence_name LIKE :prefix%d)", i)

)

kfaraz · 2023-10-23T06:17:46Z

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

-              .bind("used", usedSegments.contains(segment))
-              .bind("payload", jsonMapper.writeValueAsBytes(segment))
-              .bind("used_status_last_updated", now);
+                       .bind("id", segment.getId().toString())


Since this is a sensitive PR for a new feature, let's revert all these formatting changes for now. Better to do reformatting in refactor PRs.

Please revert all formatting changes in other lines in this class too.

Ah sorry, I must have auto-indented the whole file

kfaraz · 2023-10-23T06:21:46Z

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

+              .bind("end", interval.getEnd().toString());
+
+    for (int i = 1; i <= sequenceNamePrefixes.size(); i++) {
+      query.bind(StringUtils.format("prefix%d", i), sequenceNamePrefixes.get(i - 1) + "%");


What do we need the % for in sequenceNamePrefixes.get(i - 1) + "%"

Because we want the sequence to begin with the sequence name prefix and not just be equal to it

…5169) * Filter pending segments upgraded with transactional replace * Push sequence name filter to metadata query

…15234) * Filter pending segments upgraded with transactional replace * Push sequence name filter to metadata query

…5169) * Filter pending segments upgraded with transactional replace * Push sequence name filter to metadata query

Filter pending segments upgraded with transactional replace

37613c6

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

AmatyaAvadhanula added this to the 28.0 milestone Oct 16, 2023

kfaraz reviewed Oct 16, 2023

View reviewed changes

Push sequence name filter to metadata query

368312e

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

View reviewed changes

AmatyaAvadhanula added 6 commits October 18, 2023 09:31

Improve coverage

548c4c0

Fix tests

9120210

Fix checkstyle

c100b02

Add javadoc

156d809

Resolve merge conflicts

53d509a

Resolve conflicts

7b27f4b

kfaraz reviewed Oct 23, 2023

View reviewed changes

AmatyaAvadhanula added 2 commits October 23, 2023 13:18

Address feedback

e03329e

Revert indentation changes

daf0337

AmatyaAvadhanula requested a review from kfaraz October 23, 2023 07:58

Simplify pending segment query

92a853c

kfaraz approved these changes Oct 23, 2023

View reviewed changes

AmatyaAvadhanula merged commit 65b69cd into apache:master Oct 23, 2023
82 checks passed

AmatyaAvadhanula mentioned this pull request Oct 23, 2023

[Backport] Filter pending segments upgraded with transactional replace (#15169) #15234

Merged

LakshSingla pushed a commit that referenced this pull request Oct 23, 2023

Filter pending segments upgraded with transactional replace (#15169) (#…

701b9af

…15234) * Filter pending segments upgraded with transactional replace * Push sequence name filter to metadata query

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

Filter pending segments upgraded with transactional replace (apache#1…

0b1e2ad

…5169) * Filter pending segments upgraded with transactional replace * Push sequence name filter to metadata query

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Filter pending segments upgraded with transactional replace #15169

Filter pending segments upgraded with transactional replace #15169

AmatyaAvadhanula commented Oct 16, 2023 •

edited

Loading

kfaraz left a comment

kfaraz Oct 16, 2023

kfaraz Oct 23, 2023

AmatyaAvadhanula Oct 23, 2023

kfaraz Oct 16, 2023

AmatyaAvadhanula Oct 17, 2023

kfaraz Oct 16, 2023

AmatyaAvadhanula Oct 17, 2023

kfaraz left a comment

kfaraz Oct 20, 2023

AmatyaAvadhanula Oct 23, 2023

kfaraz Oct 23, 2023

kfaraz Oct 23, 2023

AmatyaAvadhanula Oct 23, 2023

kfaraz Oct 23, 2023

AmatyaAvadhanula Oct 23, 2023

kfaraz Oct 23, 2023

AmatyaAvadhanula Oct 23, 2023

AmatyaAvadhanula Oct 23, 2023

kfaraz Oct 23, 2023

kfaraz Oct 23, 2023

AmatyaAvadhanula Oct 23, 2023

kfaraz Oct 23, 2023

AmatyaAvadhanula Oct 23, 2023

kfaraz Oct 23, 2023

AmatyaAvadhanula Oct 23, 2023

kfaraz Oct 23, 2023

AmatyaAvadhanula Oct 23, 2023

	* @param activeBaseSequenceNames of base sequence names of active / pending completion task groups of the supervisor
	* @param activeBaseSequenceNames Set of base sequence names of active and pending completion task groups of the supervisor (if any) for this datasource

Filter pending segments upgraded with transactional replace #15169

Filter pending segments upgraded with transactional replace #15169

Conversation

AmatyaAvadhanula commented Oct 16, 2023 • edited Loading

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AmatyaAvadhanula commented Oct 16, 2023 •

edited

Loading