FrankChen021 commented on code in PR #19772:
URL: https://github.com/apache/druid/pull/19772#discussion_r3665378186
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/duty/UnusedSegmentsKiller.java:
##########
@@ -462,7 +490,7 @@ protected List<DataSegmentPlus>
fetchNextBatchOfUnusedSegments(TaskToolbox toolb
getDataSource(),
getInterval(),
getMaxUsedStatusLastUpdatedTime(),
- MAX_SEGMENTS_TO_KILL_IN_INTERVAL
+ MAX_SEGMENTS_TO_KILL_IN_BATCH
Review Comment:
[P1] Honor the candidate limit when fetching a batch
KillUnusedSegmentsTask passes a shrinking nextBatchSize to enforce
candidate.numSegmentsToKill(), but this override discards it and always fetches
1,000 rows. A candidate counted as one row by the capped discovery query can
therefore delete up to 1,000 eligible segments; across 10,000 candidates, a
cycle can delete far more than the documented 200,000 maximum. This is
reachable when the inner SQL limit samples only part of an interval or more
segments become eligible before the task runs. Pass nextBatchSize here; the
updated 200k test currently stops after queue construction and never verifies
actual deletions.
##########
server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java:
##########
@@ -1706,6 +1775,26 @@ private Interval mapToInterval(ResultSet resultSet,
String dataSource)
}
}
+ /**
+ * Reads the fields {@code dataSource}, {@code start}, {@code end} and
+ * {@code totalCount} from the given result set.
+ */
+ @Nullable
+ private Pair<DatasourceInterval, Integer>
mapToUnusedSegmentInterval(ResultSet resultSet)
+ {
+ try {
+ final String dataSource = resultSet.getString("dataSource");
+ final Interval interval = mapToInterval(resultSet, "");
+ final int totalCount = resultSet.getInt("totalCount");
+
+ return Pair.of(new DatasourceInterval(dataSource, interval), totalCount);
Review Comment:
[P2] Drop rows whose interval failed to parse
mapToInterval deliberately returns null after logging an invalid start/end
value, but this method wraps that null in a non-null DatasourceInterval. The
later filter therefore retains it, and UnusedSegmentsKiller dereferences
entry.interval() while ordering or emitting metrics, aborting the entire queue
rebuild because of one malformed metadata row. Return null when interval is
null so the existing filter actually skips the bad row.
##########
server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java:
##########
@@ -1116,6 +1118,70 @@ public List<Interval>
retrieveSomeUnusedSegmentIntervals(String dataSource, int
return
intervals.stream().filter(Objects::nonNull).collect(Collectors.toList());
}
+ /**
+ * Scans upto {@code maxSegmentsToScan} unused segments which are eligible
for
+ * kill and returns the unique datasource-interval for the segments scanned.
+ * <p>
+ * This method ensures that if there is any unused segment in any datasource
+ * which was updated earlier than {@code maxUpdatedTime}, then the returned
+ * map is not empty. However, it does NOT guarantee that:
+ * <ul>
+ * <li>the candidates in the returned map would be ordered by datasource or
interval</li>
+ * <li>the result would contain {@code limit} entries when there are more
distinct
+ * intervals with eligible unused segments in the metadata store.</li>
+ * </ul>
+ *
+ * @param maxUpdatedTime Unused segments are considered eligible for kill
+ * if they were last updated before this time.
+ * @param maxResultSize Maximum number of candidate intervals to return
+ * across all datasources.
+ * @param maxSegmentsToScan Maximum number of eligible unused segments to
scan
+ * in the metadata store.
+ * @return Map from {@link DatasourceInterval} to the number of unused
segments
+ * eligible for kill.
+ */
+ public Map<DatasourceInterval, Integer> retrieveSomeUnusedSegmentIntervals(
+ DateTime maxUpdatedTime,
+ int maxResultSize,
+ int maxSegmentsToScan
+ )
+ {
+ final String sql = StringUtils.format(
+ // Disable checkstyle to avoid argumentLineBreaking rule from getting
triggered
+ //CHECKSTYLE.OFF: Regexp
+ """
+ SELECT dataSource, start, %2$send%2$s, COUNT(*) AS totalCount
+ FROM (
+ SELECT dataSource, start, %2$send%2$s
+ FROM %1$s
+ WHERE used = false
+ AND (used_status_last_updated IS NULL OR
used_status_last_updated <= :maxUpdatedTime)
+ %3$s
+ ) AS unused
+ GROUP BY dataSource, %2$send%2$s, start
+ %4$s
+ """,
+ //CHECKSTYLE.ON: Regexp
+ dbTables.getSegmentsTable(),
+ connector.getQuoteString(),
+ connector.limitClause(maxSegmentsToScan),
Review Comment:
[P1] Prevent locked rows from monopolizing the global scan
The new global LIMIT has no ordering, cursor, or rotation state. If the
selected 200,000 rows belong to intervals whose exclusive locks cannot be
acquired, all generated tasks are skipped, the rows remain, and the next
rebuild can select the identical slice again. Eligible segments outside that
slice—including unrelated datasources—can then be starved indefinitely, whereas
the previous per-datasource discovery still allowed other datasources to
progress. Candidate discovery needs a deterministic progress/fairness mechanism
that can advance past repeatedly skipped intervals.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
