Compaction: Fetch segments one at a time on main task; skip when possible.

[gianm]

Compact tasks include the ability to fetch existing segments and determine
reasonable defaults for granularitySpec, dimensionsSpec, and metricsSpec.
This is a useful feature that makes compact tasks work well even when the
user running the compaction does not have a clear idea of what they want
the compacted segments to be like.

However, this comes at a cost: it takes time, and disk space, to do all
of these fetches. This patch improves the situation in two ways:

1. When segments do need to be fetched, download them one at a time and
   delete them when we're done. This still takes time, but minimizes the
   required disk space.

2. Don't fetch segments on the main compact task when they aren't needed.
   If the user provides a full granularitySpec, dimensionsSpec, and
   metricsSpec, we can skip it.

[kfaraz]

Looks good, added minor style suggestions.
Since the download of segments was already sequential, this shouldn't negatively impact the task run time either.

[kfaraz]

Thank you! The whole thing is much easier to read now after removing the Pairs.

[kfaraz]

This can probably be simplified by having a single boolean rollup member field.

The field is initialized to true.

Then, this whole block becomes:

Suggested change

	if (rollupIsValid) {
	Boolean isRollup = index.getMetadata().isRollup();
	if (isRollup == null) {
	rollupIsValid = false;
	rollup = false;
	} else if (rollup == null) {
	rollup = isRollup;
	} else if (!isRollup.equals(rollup)) {
	rollupIsValid = false;
	rollup = false;
	}
	}
	Boolean isIndexRollup = index.getMetadata().isRollup();
	rollup = rollup && isIndexRollup != null && isIndexRollup.equals(rollup)

[gianm]

Good idea. Updated to similar but equivalent:

final Boolean isIndexRollup = index.getMetadata().isRollup();
rollup = rollup && Boolean.valueOf(true).equals(isIndexRollup);

[kfaraz]

Would this work?

Suggested change

	(o1, o2) -> Comparators.intervalsByStartThenEnd().compare(o2.lhs.getInterval(), o1.lhs.getInterval())
	Comparators.intervalsByStartThenEnd().reversed()

[gianm]

Something still needs to extract the interval. I changed it to this to try to improve clarity:

      segments.sort(
          Comparator.comparing(
              o -> o.lhs.getInterval(),
              Comparators.intervalsByStartThenEnd().reversed()
          )
      );

[kfaraz]

Nit:

Suggested change

	private void sortSegmentsList()
	private void sortSegmentsListNewestFirst()

[kfaraz]

I wonder if it would be cleaner to pass all these specs to the ExistingSegmentAnalyzer and let it decide whether it needs the spec or not. This way, only the ExistingSegmentAnalyzer would be doing the null checks.
But maybe this is better as ExistingSegmentAnalyzer only has to deal with the specs of the downloaded segments.

[gianm]

"ExistingSegmentAnalyzer only has to deal with the specs of the downloaded segments" is what I was thinking. I wanted to pass in the minimal amount of information to ExistingSegmentAnalyzer, in order to keep its logic as simple as possible.

[maytasm]

@gianm What do you think of a option for the Compact task to fetch dimensionsSpec, metricsSpec, granularitySpec from the latest supervisor spec? (obviously only for datasource that was ingested via streaming ingestion)