maytasm commented on a change in pull request #10371:
URL: https://github.com/apache/druid/pull/10371#discussion_r488385897
##########
File path:
server/src/main/java/org/apache/druid/server/coordinator/duty/NewestSegmentFirstIterator.java
##########
@@ -112,27 +114,38 @@
}
@Override
- public Object2LongOpenHashMap<String> totalRemainingSegmentsSizeBytes()
+ public Map<String, CompactionStatistics> totalRemainingStatistics()
{
- final Object2LongOpenHashMap<String> resultMap = new
Object2LongOpenHashMap<>();
- resultMap.defaultReturnValue(UNKNOWN_TOTAL_REMAINING_SEGMENTS_SIZE);
- for (QueueEntry entry : queue) {
- final VersionedIntervalTimeline<String, DataSegment> timeline =
dataSources.get(entry.getDataSource());
- final Interval interval = new
Interval(timeline.first().getInterval().getStart(), entry.interval.getEnd());
-
- final List<TimelineObjectHolder<String, DataSegment>> holders =
timeline.lookup(interval);
-
- long size = 0;
- for (DataSegment segment : FluentIterable
- .from(holders)
- .transformAndConcat(TimelineObjectHolder::getObject)
- .transform(PartitionChunk::getObject)) {
- size += segment.getSize();
- }
+ return remainingSegments;
+ }
+
+ @Override
+ public Map<String, CompactionStatistics> totalProcessedStatistics()
+ {
+ return processedSegments;
+ }
- resultMap.put(entry.getDataSource(), size);
+ @Override
+ public void flushAllSegments()
+ {
+ if (queue.isEmpty()) {
+ return;
+ }
+ QueueEntry entry;
+ while ((entry = queue.poll()) != null) {
+ final List<DataSegment> resultSegments = entry.segments;
+ final String dataSourceName = resultSegments.get(0).getDataSource();
+ // This entry was in the queue, meaning that it was not processed.
Hence, also aggregates it's
+ // statistic to the remaining segments counts.
+ collectSegmentStatistics(remainingSegments, dataSourceName, new
SegmentsToCompact(entry.segments));
+ final CompactibleTimelineObjectHolderCursor
compactibleTimelineObjectHolderCursor = timelineIterators.get(
+ dataSourceName
+ );
+ // WARNING: This iterates the compactibleTimelineObjectHolderCursor.
+ // Since this method is intended to only be use after all necessary
iteration is done on this iterator
Review comment:
I mean that this method (`flushAllSegments`), will iterates the
`compactibleTimelineObjectHolderCursor` in `iterateAllSegments` method call.
This class `NewestSegmentFirstIterator` when use as a iterator in the next()
method will also iterate the `compactibleTimelineObjectHolderCursor`.
Hence, this iterator (`NewestSegmentFirstIterator) cannot be use to iterate
after this method (`flushAllSegments`) is called.
Basically, you cannot call `flushAllSegments` while iterating the
NewestSegmentFirstIterator and you cannot call `flushAllSegments` then go back
to iterating the NewestSegmentFirstIterator
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org
For additional commands, e-mail: commits-h...@druid.apache.org
