capistrant commented on code in PR #20805:
URL: https://github.com/apache/druid/pull/20805#discussion_r4469742896
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ReadableInputQueue.java:
##########
@@ -178,42 +201,42 @@ public ReadableInputQueue(
readablePartitions.add(new NumberedItem<>(inputNumber, partition));
}
}
+
+ if (timeOrder != Order.NONE) {
+ // Stable sort: segments with equal intervals keep their original order.
+ segments.sort(Comparator.comparing(item ->
item.item().descriptor().getInterval(), intervalOrder));
+ }
+ loadableSegments.addAll(segments);
}
/**
* If this method has not yet been called, then:
- * (1) transition all locally-cached segments out of {@link
#loadableSegments}
+ * (1) when {@link #timeOrder} is {@link Order#NONE}, transition all
locally-cached segments from
+ * {@link #loadableSegments} to {@link #pendingNextInputs}, so they are
handed out first. Otherwise, cached segments
+ * keep their place in time order.
* (2) start loading up to {@link #loadahead} additional segments for future
calls to {@link #nextInput()}
* If this method has previously been called, subsequent calls do nothing.
* This is separated from the constructor because we don't want to acquire
resources immediately on construction.
*/
public void start()
{
if (started.compareAndSet(false, true)) {
- // (1) acquire all locally-cached segments
- synchronized (this) {
- // Temporarily store all non-cached segments.
- final List<NumberedItem<LoadableSegment>> toLoad = new ArrayList<>();
- NumberedItem<LoadableSegment> nextItem;
- while ((nextItem = loadableSegments.poll()) != null) {
- final LoadableSegment loadableSegment = nextItem.item();
- final Optional<Segment> cachedSegment =
loadableSegment.acquireIfCached(acquireMode);
- if (cachedSegment.isPresent()) {
- final SegmentReferenceHolder holder = new SegmentReferenceHolder(
- new SegmentReference(loadableSegment.descriptor(),
cachedSegment),
- loadableSegment.description()
- );
- loadedSegments.add(holder);
- pendingNextInputs.add(
- Futures.immediateFuture(
- new NumberedReadableInput(nextItem.inputNumber(),
ReadableInput.segment(holder))
- )
- );
- } else {
- toLoad.add(nextItem);
+ // (1) acquire all locally-cached segments if timeOrder is NONE,
otherwise acquire them in time order
+ if (timeOrder == Order.NONE) {
Review Comment:
good point.
> Is it possible to pull cached segments to the front of the line whenever
we enter a new time chunk
this looks feasible, will work on it today
--
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]
