You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org> on 2023/10/16 14:37:46 UTC
[PR] Filter pending segments upgraded with transactional replace (druid)
AmatyaAvadhanula opened a new pull request, #15169:
URL: https://github.com/apache/druid/pull/15169
The pending segment upgrade introduced as part of https://github.com/apache/druid/pull/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.
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
This PR has:
- [x] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [x] been tested in a test Druid cluster.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368463676
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
+ Map<SegmentIdWithShardSpec, String> getPendingSegmentsForIntervalWithHandle(
+ final Handle handle,
+ final String dataSource,
+ final Interval interval,
+ final Set<String> sequenceNamePrefixFilter
+ ) throws IOException
+ {
+ if (sequenceNamePrefixFilter.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final List<String> sequenceNamePrefixes = new ArrayList<>(sequenceNamePrefixFilter);
+ StringBuilder sql = new StringBuilder(
+ "SELECT sequence_name, payload FROM "
+ + dbTables.getPendingSegmentsTable()
+ + " WHERE dataSource = :dataSource AND start < :end AND "
+ + connector.getQuoteString() + "end" + connector.getQuoteString() + " > :start"
+ );
+
+ sql.append(" AND ( ");
+
+ 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(")");
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368236302
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
+ Map<SegmentIdWithShardSpec, String> getPendingSegmentsForIntervalWithHandle(
+ final Handle handle,
+ final String dataSource,
+ final Interval interval,
+ final Set<String> sequenceNamePrefixFilter
+ ) throws IOException
+ {
+ if (sequenceNamePrefixFilter.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final List<String> sequenceNamePrefixes = new ArrayList<>(sequenceNamePrefixFilter);
+ StringBuilder sql = new StringBuilder(
+ "SELECT sequence_name, payload FROM "
+ + dbTables.getPendingSegmentsTable()
+ + " WHERE dataSource = :dataSource AND start < :end AND "
+ + connector.getQuoteString() + "end" + connector.getQuoteString() + " > :start"
+ );
+
+ sql.append(" AND ( ");
+
+ 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(")");
Review Comment:
There needs to be an OR in between the first size - 1 conditions and the immediately next ones. Did you mean that it's better to use 0 indexing rather than 1 based?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368253576
##########
extensions-contrib/materialized-view-maintenance/src/main/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisor.java:
##########
@@ -295,6 +295,12 @@ public LagStats computeLagStats()
throw new UnsupportedOperationException("Compute Lag Stats not supported in MaterializedViewSupervisor");
}
+ @Override
+ public Set<String> getActiveBaseSequenceNames()
+ {
+ throw new UnsupportedOperationException("Get Active sequence names is not supported in MaterializedViewSupervisor");
Review Comment:
Fixed. Missed this earlier
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "github-code-scanning[bot] (via GitHub)" <gi...@apache.org>.
github-code-scanning[bot] commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1362450549
##########
server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java:
##########
@@ -464,6 +466,44 @@
);
}
+ private Boolean insertPendingSegmentAndSequenceName(Pair<SegmentIdWithShardSpec, String> pendingSegmentSequenceName)
+ {
+ final SegmentIdWithShardSpec pendingSegment = pendingSegmentSequenceName.lhs;
+ final String sequenceName = pendingSegmentSequenceName.rhs;
+ final String table = derbyConnectorRule.metadataTablesConfigSupplier().get().getPendingSegmentsTable();
+ return derbyConnector.retryWithHandle(
+ handle -> {
+ handle.createStatement(
+ StringUtils.format(
+ "INSERT INTO %1$s (id, dataSource, created_date, start, %2$send%2$s, sequence_name, sequence_prev_id, "
+ + "sequence_name_prev_id_sha1, payload) "
+ + "VALUES (:id, :dataSource, :created_date, :start, :end, :sequence_name, :sequence_prev_id, "
+ + ":sequence_name_prev_id_sha1, :payload)",
+ table,
+ derbyConnector.getQuoteString()
+ )
+ )
+ .bind("id", pendingSegment.toString())
+ .bind("dataSource", pendingSegment.getDataSource())
+ .bind("created_date", DateTimes.nowUtc().toString())
+ .bind("start", pendingSegment.getInterval().getStart().toString())
+ .bind("end", pendingSegment.getInterval().getEnd().toString())
+ .bind("sequence_name", sequenceName)
+ .bind("sequence_prev_id", pendingSegment.toString())
+ .bind("sequence_name_prev_id_sha1", BaseEncoding.base16().encode(
+ Hashing.sha1()
Review Comment:
## Deprecated method or constructor invocation
Invoking [Hashing.sha1](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/5912)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368230121
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
Review Comment:
The method invoking this internally would ignore segments that are not needed eventually, and it would be hard to test this optimization.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368236302
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
+ Map<SegmentIdWithShardSpec, String> getPendingSegmentsForIntervalWithHandle(
+ final Handle handle,
+ final String dataSource,
+ final Interval interval,
+ final Set<String> sequenceNamePrefixFilter
+ ) throws IOException
+ {
+ if (sequenceNamePrefixFilter.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final List<String> sequenceNamePrefixes = new ArrayList<>(sequenceNamePrefixFilter);
+ StringBuilder sql = new StringBuilder(
+ "SELECT sequence_name, payload FROM "
+ + dbTables.getPendingSegmentsTable()
+ + " WHERE dataSource = :dataSource AND start < :end AND "
+ + connector.getQuoteString() + "end" + connector.getQuoteString() + " > :start"
+ );
+
+ sql.append(" AND ( ");
+
+ 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(")");
Review Comment:
There needs to be an OR in between. Did you mean that it's better to use 0 indexing rather than 1 based?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368254389
##########
server/src/main/java/org/apache/druid/indexing/overlord/supervisor/Supervisor.java:
##########
@@ -93,4 +94,9 @@ default Boolean isHealthy()
LagStats computeLagStats();
int getActiveTaskGroupsCount();
+
+ /**
+ * @return active base sequence names for reading and pending completion task groups of a seekable stream supervisor
Review Comment:
Done
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
+ Map<SegmentIdWithShardSpec, String> getPendingSegmentsForIntervalWithHandle(
+ final Handle handle,
+ final String dataSource,
+ final Interval interval,
+ final Set<String> sequenceNamePrefixFilter
+ ) throws IOException
+ {
+ if (sequenceNamePrefixFilter.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final List<String> sequenceNamePrefixes = new ArrayList<>(sequenceNamePrefixFilter);
+ StringBuilder sql = new StringBuilder(
Review Comment:
Done. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368238661
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -1833,26 +1900,26 @@ private Set<DataSegment> announceHistoricalSegmentBatch(
for (DataSegment segment : partition) {
final String now = DateTimes.nowUtc().toString();
preparedBatch.add()
- .bind("id", segment.getId().toString())
- .bind("dataSource", segment.getDataSource())
- .bind("created_date", now)
- .bind("start", segment.getInterval().getStart().toString())
- .bind("end", segment.getInterval().getEnd().toString())
- .bind("partitioned", (segment.getShardSpec() instanceof NoneShardSpec) ? false : true)
- .bind("version", segment.getVersion())
- .bind("used", usedSegments.contains(segment))
- .bind("payload", jsonMapper.writeValueAsBytes(segment))
- .bind("used_status_last_updated", now);
+ .bind("id", segment.getId().toString())
Review Comment:
Ah sorry, I must have auto-indented the whole file
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368340709
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
Review Comment:
Sorry, It seems like it filters segments that have already been upgraded, and not all the unnecessary segments. I can make this change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1363270146
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1093,6 +1093,21 @@ public void resetOffsets(@Nonnull DataSourceMetadata resetDataSourceMetadata)
addNotice(new ResetOffsetsNotice(resetDataSourceMetadata));
}
+ @Override
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368165054
##########
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
Review Comment:
Please update the javadoc to say prefixes instead of sequence names.
##########
extensions-contrib/materialized-view-maintenance/src/test/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisorSpecTest.java:
##########
@@ -217,6 +217,13 @@ public void testMaterializedViewSupervisorSpecCreated()
Assert.assertTrue(e instanceof UnsupportedOperationException);
}
+ try {
Review Comment:
Use `Assert.assertThrows` instead and update the existing assertions in this method too.
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
Review Comment:
Avoid using this. It is okay to just test the method which invokes this internally.
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -1833,26 +1900,26 @@ private Set<DataSegment> announceHistoricalSegmentBatch(
for (DataSegment segment : partition) {
final String now = DateTimes.nowUtc().toString();
preparedBatch.add()
- .bind("id", segment.getId().toString())
- .bind("dataSource", segment.getDataSource())
- .bind("created_date", now)
- .bind("start", segment.getInterval().getStart().toString())
- .bind("end", segment.getInterval().getEnd().toString())
- .bind("partitioned", (segment.getShardSpec() instanceof NoneShardSpec) ? false : true)
- .bind("version", segment.getVersion())
- .bind("used", usedSegments.contains(segment))
- .bind("payload", jsonMapper.writeValueAsBytes(segment))
- .bind("used_status_last_updated", now);
+ .bind("id", segment.getId().toString())
Review Comment:
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.
##########
server/src/main/java/org/apache/druid/indexing/overlord/supervisor/Supervisor.java:
##########
@@ -93,4 +94,9 @@ default Boolean isHealthy()
LagStats computeLagStats();
int getActiveTaskGroupsCount();
+
+ /**
+ * @return active base sequence names for reading and pending completion task groups of a seekable stream supervisor
Review Comment:
Update: prefixes instead of sequence names.
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
+ Map<SegmentIdWithShardSpec, String> getPendingSegmentsForIntervalWithHandle(
+ final Handle handle,
+ final String dataSource,
+ final Interval interval,
+ final Set<String> sequenceNamePrefixFilter
+ ) throws IOException
+ {
+ if (sequenceNamePrefixFilter.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final List<String> sequenceNamePrefixes = new ArrayList<>(sequenceNamePrefixFilter);
+ StringBuilder sql = new StringBuilder(
+ "SELECT sequence_name, payload FROM "
+ + dbTables.getPendingSegmentsTable()
+ + " WHERE dataSource = :dataSource AND start < :end AND "
+ + connector.getQuoteString() + "end" + connector.getQuoteString() + " > :start"
+ );
+
+ sql.append(" AND ( ");
+
+ 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(")");
+
+ sql.append(" )");
+
+ Query<Map<String, Object>> query =
+ handle.createQuery(sql.toString())
+ .bind("dataSource", dataSource)
+ .bind("start", interval.getStart().toString())
+ .bind("end", interval.getEnd().toString());
+
+ for (int i = 1; i <= sequenceNamePrefixes.size(); i++) {
+ query.bind(StringUtils.format("prefix%d", i), sequenceNamePrefixes.get(i - 1) + "%");
Review Comment:
What do we need the `%` for in `sequenceNamePrefixes.get(i - 1) + "%"`
##########
extensions-contrib/materialized-view-maintenance/src/main/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisor.java:
##########
@@ -295,6 +295,12 @@ public LagStats computeLagStats()
throw new UnsupportedOperationException("Compute Lag Stats not supported in MaterializedViewSupervisor");
}
+ @Override
+ public Set<String> getActiveBaseSequenceNames()
+ {
+ throw new UnsupportedOperationException("Get Active sequence names is not supported in MaterializedViewSupervisor");
Review Comment:
Please fix this as the message is redundant.
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
+ Map<SegmentIdWithShardSpec, String> getPendingSegmentsForIntervalWithHandle(
+ final Handle handle,
+ final String dataSource,
+ final Interval interval,
+ final Set<String> sequenceNamePrefixFilter
+ ) throws IOException
+ {
+ if (sequenceNamePrefixFilter.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final List<String> sequenceNamePrefixes = new ArrayList<>(sequenceNamePrefixFilter);
+ StringBuilder sql = new StringBuilder(
Review Comment:
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 `OR`s.
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
+ Map<SegmentIdWithShardSpec, String> getPendingSegmentsForIntervalWithHandle(
+ final Handle handle,
+ final String dataSource,
+ final Interval interval,
+ final Set<String> sequenceNamePrefixFilter
+ ) throws IOException
+ {
+ if (sequenceNamePrefixFilter.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final List<String> sequenceNamePrefixes = new ArrayList<>(sequenceNamePrefixFilter);
+ StringBuilder sql = new StringBuilder(
Review Comment:
Use `.append` instead of `+` as it is already a `StringBuilder`.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1093,6 +1093,21 @@ public void resetOffsets(@Nonnull DataSourceMetadata resetDataSourceMetadata)
addNotice(new ResetOffsetsNotice(resetDataSourceMetadata));
}
+ @Override
Review Comment:
Thanks!
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
+ Map<SegmentIdWithShardSpec, String> getPendingSegmentsForIntervalWithHandle(
+ final Handle handle,
+ final String dataSource,
+ final Interval interval,
+ final Set<String> sequenceNamePrefixFilter
+ ) throws IOException
+ {
+ if (sequenceNamePrefixFilter.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final List<String> sequenceNamePrefixes = new ArrayList<>(sequenceNamePrefixFilter);
+ StringBuilder sql = new StringBuilder(
+ "SELECT sequence_name, payload FROM "
+ + dbTables.getPendingSegmentsTable()
+ + " WHERE dataSource = :dataSource AND start < :end AND "
+ + connector.getQuoteString() + "end" + connector.getQuoteString() + " > :start"
+ );
+
+ sql.append(" AND ( ");
+
+ 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(")");
Review Comment:
Cleaner to just do this `for i = 0 to sequenceNamePrefixes.size()`:
```suggestion
sequenceFilterCondition.append(
StringUtils.format("(sequence_name LIKE :prefix%d)", i)
)
```
##########
indexing-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)) {
Review Comment:
Please invert the if condition:
```
if contains:
return value
else
return empty
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368253922
##########
extensions-contrib/materialized-view-maintenance/src/test/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisorSpecTest.java:
##########
@@ -217,6 +217,13 @@ public void testMaterializedViewSupervisorSpecCreated()
Assert.assertTrue(e instanceof UnsupportedOperationException);
}
+ try {
Review Comment:
Done
##########
indexing-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)) {
Review Comment:
Done
##########
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
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368236302
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
+ Map<SegmentIdWithShardSpec, String> getPendingSegmentsForIntervalWithHandle(
+ final Handle handle,
+ final String dataSource,
+ final Interval interval,
+ final Set<String> sequenceNamePrefixFilter
+ ) throws IOException
+ {
+ if (sequenceNamePrefixFilter.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final List<String> sequenceNamePrefixes = new ArrayList<>(sequenceNamePrefixFilter);
+ StringBuilder sql = new StringBuilder(
+ "SELECT sequence_name, payload FROM "
+ + dbTables.getPendingSegmentsTable()
+ + " WHERE dataSource = :dataSource AND start < :end AND "
+ + connector.getQuoteString() + "end" + connector.getQuoteString() + " > :start"
+ );
+
+ sql.append(" AND ( ");
+
+ 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(")");
Review Comment:
There needs to be an OR in between the first size - 1 conditions and the immediately next ones. Did you mean that it's better to use 0 indexing rather than 1 based?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1362357984
##########
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;
Review Comment:
Done
##########
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
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1361105395
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1093,6 +1093,21 @@ public void resetOffsets(@Nonnull DataSourceMetadata resetDataSourceMetadata)
addNotice(new ResetOffsetsNotice(resetDataSourceMetadata));
}
+ @Override
Review Comment:
Would be nice to add a javadoc here or in the interface to say a few words about the base sequence name and what kind of strings we can expect in it.
You could also link to the javadoc of `TaskGroup.baseSequenceName` if one already exists. If it doesn't, maybe add a getter for this field `TaskGroup.getBaseSequenceName()` and write a small javadoc for it.
##########
extensions-contrib/materialized-view-maintenance/src/main/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisor.java:
##########
@@ -295,6 +295,12 @@ public LagStats computeLagStats()
throw new UnsupportedOperationException("Compute Lag Stats not supported in MaterializedViewSupervisor");
}
+ @Override
+ public Set<String> getActiveBaseSequenceNames()
+ {
+ throw new UnsupportedOperationException("Get Active sequence names is not supported in MaterializedViewSupervisor");
Review Comment:
Nit: Probably don't need an explicit message here.
##########
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
Review Comment:
```suggestion
* @param activeBaseSequenceNames Set of base sequence names of active and pending completion task groups of the supervisor (if any) for this datasource
```
##########
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;
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula commented on code in PR #15169:
URL: https://github.com/apache/druid/pull/15169#discussion_r1368231397
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -258,9 +258,73 @@ public int markSegmentsAsUnusedWithinInterval(String dataSource, Interval interv
/**
* Fetches all the pending segments, whose interval overlaps with the given
- * search interval from the metadata store. Returns a Map from the
- * pending segment ID to the sequence name.
+ * search interval and has a sequence_name that begins with one of the prefixes in sequenceNamePrefixFilter
+ * from the metadata store. Returns a Map from the pending segment ID to the sequence name.
*/
+ @VisibleForTesting
+ Map<SegmentIdWithShardSpec, String> getPendingSegmentsForIntervalWithHandle(
+ final Handle handle,
+ final String dataSource,
+ final Interval interval,
+ final Set<String> sequenceNamePrefixFilter
+ ) throws IOException
+ {
+ if (sequenceNamePrefixFilter.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final List<String> sequenceNamePrefixes = new ArrayList<>(sequenceNamePrefixFilter);
+ StringBuilder sql = new StringBuilder(
+ "SELECT sequence_name, payload FROM "
+ + dbTables.getPendingSegmentsTable()
+ + " WHERE dataSource = :dataSource AND start < :end AND "
+ + connector.getQuoteString() + "end" + connector.getQuoteString() + " > :start"
+ );
+
+ sql.append(" AND ( ");
+
+ 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(")");
+
+ sql.append(" )");
+
+ Query<Map<String, Object>> query =
+ handle.createQuery(sql.toString())
+ .bind("dataSource", dataSource)
+ .bind("start", interval.getStart().toString())
+ .bind("end", interval.getEnd().toString());
+
+ for (int i = 1; i <= sequenceNamePrefixes.size(); i++) {
+ query.bind(StringUtils.format("prefix%d", i), sequenceNamePrefixes.get(i - 1) + "%");
Review Comment:
Because we want the sequence to begin with the sequence name prefix and not just be equal to it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Filter pending segments upgraded with transactional replace (druid)
Posted by "AmatyaAvadhanula (via GitHub)" <gi...@apache.org>.
AmatyaAvadhanula merged PR #15169:
URL: https://github.com/apache/druid/pull/15169
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org