iter

Author: martijnvg

x-pack/plugin/downsample/src/main/java/org/elasticsearch/xpack/downsample/DownsampleShardIndexer.java

@@ -374,7 +374,9 @@ public LeafBucketCollector getLeafCollector(final AggregationExecutionContext ag
  }
 
  void bulkCollection() throws IOException {
- for (LeafDownsampleCollector leafBucketCollector : leafBucketCollectors.reversed()) {
+ // The leaf bucked collectors newer timestamp go first, other we capture the incorrect last value for counters and labels.
+ leafBucketCollectors.sort((o1, o2) -> -Long.compare(o1.firstTimeStampForBulkCollection, o2.firstTimeStampForBulkCollection));
+ for (LeafDownsampleCollector leafBucketCollector : leafBucketCollectors) {
  leafBucketCollector.leafBulkCollection();
  }
  }
@@ -386,6 +388,8 @@ class LeafDownsampleCollector extends LeafBucketCollector {
  final FormattedDocValues[] formattedDocValues;
  final AbstractDownsampleFieldProducer[] fieldProducers;
 
+ // Capture the first timestamp in order to determine which leaf collector's leafBulkCollection() is invoked first.
+ long firstTimeStampForBulkCollection;
  final IntArrayList buffer = new IntArrayList(DOCID_BUFFER_SIZE);
  final long timestampBoundStartTime = searchExecutionContext.getIndexSettings().getTimestampBounds().startTime();
 
@@ -462,6 +466,11 @@ public void collect(int docId, long owningBucketOrd) throws IOException {
  }
  bucketsCreated++;
  }
+
+ if (buffer.isEmpty()) {
+ firstTimeStampForBulkCollection = aggCtx.getTimestamp();
+ }
+ // buffer.add() always delegates to system.arraycopy() and checks buffer size for resizing purposes:
  buffer.buffer[buffer.elementsCount++] = docId;
  if (buffer.size() == DOCID_BUFFER_SIZE) {
  bulkCollection();