server/src/main/java/org/apache/druid/server/coordinator/duty/NewestSegmentFirstIterator.java
/**
   * Find segments to compact together for the given intervalToSearch. It progressively searches the given
   * intervalToSearch in time order (latest first). The timeline lookup duration is one day. It means, the timeline is
   * looked up for the last one day of the given intervalToSearch, and the next day is searched again if the size of
   * found segments are not enough to compact. This is repeated until enough amount of segments are found.
   *
   * @return segments to compact
   */
Find segments to compact together for the given intervalToSearch. It progressively searches the given
intervalToSearch in time order (latest first). The timeline lookup duration is one day. It means, the timeline is
looked up for the last one day of the given intervalToSearch, and the next day is searched again if the size of
found segments are not enough to compact. This is repeated until enough amount of segments are found.

@return segments to compact
Find segments to compact together for the given intervalToSearch . It progressively searches the given intervalToSearch in time order ( latest first ) . The timeline lookup duration is one day . It means the timeline is looked up for the last one day of the given intervalToSearch and the next day is searched again if the size of found segments are not enough to compact . This is repeated until enough amount of segments are found . @return segments to compact
private SegmentsToCompact findSegmentsToCompact(
      final String dataSourceName,
      final CompactibleTimelineObjectHolderCursor compactibleTimelineObjectHolderCursor,
      final DataSourceCompactionConfig config
  )
  {
    final long inputSegmentSize = config.getInputSegmentSizeBytes();
    while (compactibleTimelineObjectHolderCursor.hasNext()) {
      List<DataSegment> segments = compactibleTimelineObjectHolderCursor.next();
      final SegmentsToCompact candidates = new SegmentsToCompact(segments);
      if (!candidates.isEmpty()) {
        final boolean isCompactibleSize = candidates.getTotalSize() <= inputSegmentSize;
        final boolean needsCompaction = needsCompaction(
            config,
            candidates
        );
        if (isCompactibleSize && needsCompaction) {
          if (config.getGranularitySpec() != null && config.getGranularitySpec().getSegmentGranularity() != null) {
            Interval interval = candidates.getUmbrellaInterval();
            Set<Interval> intervalsCompacted = intervalCompactedForDatasource.computeIfAbsent(dataSourceName, k -> new HashSet<>());
            if (intervalsCompacted.contains(interval)) {
              continue;
            }
            intervalsCompacted.add(interval);
          }
          return candidates;
        } else {
          if (!needsCompaction) {
            collectSegmentStatistics(compactedSegments, dataSourceName, candidates);
          } else {
            collectSegmentStatistics(skippedSegments, dataSourceName, candidates);
            log.warn(
                "total segment size[%d] for datasource[%s] and interval[%s] is larger than inputSegmentSize[%d]."
                + " Continue to the next interval.",
                candidates.getTotalSize(),
                candidates.segments.get(0).getDataSource(),
                candidates.segments.get(0).getInterval(),
                inputSegmentSize
            );
          }
        }
      } else {
        throw new ISE("No segment is found?");
      }
    }
    log.info("All segments look good! Nothing to compact");
    return new SegmentsToCompact();
  }
private SegmentsToCompact findSegmentsToCompact ( final String dataSourceName , final CompactibleTimelineObjectHolderCursor compactibleTimelineObjectHolderCursor , final DataSourceCompactionConfig config ) { final long inputSegmentSize = config . getInputSegmentSizeBytes ( ) ; while ( compactibleTimelineObjectHolderCursor . hasNext ( ) ) { List < DataSegment > segments = compactibleTimelineObjectHolderCursor . next ( ) ; final SegmentsToCompact candidates = new SegmentsToCompact ( segments ) ; if ( ! candidates . isEmpty ( ) ) { final boolean isCompactibleSize = candidates . getTotalSize ( ) <= inputSegmentSize ; final boolean needsCompaction = needsCompaction ( config , candidates ) ; if ( isCompactibleSize && needsCompaction ) { if ( config . getGranularitySpec ( ) != null && config . getGranularitySpec ( ) . getSegmentGranularity ( ) != null ) { Interval interval = candidates . getUmbrellaInterval ( ) ; Set < Interval > intervalsCompacted = intervalCompactedForDatasource . computeIfAbsent ( dataSourceName , k -> new HashSet < > ( ) ) ; if ( intervalsCompacted . contains ( interval ) ) { continue ; } intervalsCompacted . add ( interval ) ; } return candidates ; } else { if ( ! needsCompaction ) { collectSegmentStatistics ( compactedSegments , dataSourceName , candidates ) ; } else { collectSegmentStatistics ( skippedSegments , dataSourceName , candidates ) ; log . warn ( "total segment size[%d] for datasource[%s] and interval[%s] is larger than inputSegmentSize[%d]." + " Continue to the next interval." , candidates . getTotalSize ( ) , candidates . segments . get ( 0 ) . getDataSource ( ) , candidates . segments . get ( 0 ) . getInterval ( ) , inputSegmentSize ) ; } } } else { throw new ISE ( "No segment is found?" ) ; } } log . info ( "All segments look good! Nothing to compact" ) ; return new SegmentsToCompact ( ) ; }
Find segments to compact together for the given intervalToSearch.
Find segments to compact together for the given intervalToSearch .
// Skip this candidates if we have compacted the interval already // Collect statistic for segments that is already compacted // Collect statistic for segments that is skipped // Note that if segments does not need compaction then we do not double count here