processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/StreamingMergeSortedGrouper.java
/**
   * Return a sorted iterator.  This method can be called safely while writing, and the iterating thread and the writing
   * thread can be different.  The result iterator always returns sorted results.  This method should be called only one
   * time per grouper.
   *
   * @return a sorted iterator
   */
Return a sorted iterator.  This method can be called safely while writing, and the iterating thread and the writing
thread can be different.  The result iterator always returns sorted results.  This method should be called only one
time per grouper.

@return a sorted iterator
Return a sorted iterator . This method can be called safely while writing and the iterating thread and the writing thread can be different . The result iterator always returns sorted results . This method should be called only one time per grouper . @return a sorted iterator
public CloseableIterator<Entry<KeyType>> iterator()
  {
    if (!initialized) {
      throw new ISE("Grouper should be initialized first");
    }
    return new CloseableIterator<Entry<KeyType>>()
    {
      final ReusableEntry<KeyType> reusableEntry = ReusableEntry.create(keySerde, aggregators.length);
      {
        increaseReadIndexTo(0);
      }
      @Override
      public boolean hasNext()
      {
         The below line can be executed between increasing curWriteIndex and setting finished in
         StreamingMergeSortedGrouper.finish(), but it is also a valid case because there should be at least one slot
         which is not read yet before finished is set.
        return !finished || remaining() > 0;
      }
      /**
       * Calculate the number of remaining items in the array.  Must be called only when
       * {@link StreamingMergeSortedGrouper#finished} is true.
       *
       * @return the number of remaining items
       */
      private int remaining()
      {
        if (curWriteIndex >= nextReadIndex) {
          return curWriteIndex - nextReadIndex;
        } else {
          return (maxNumSlots - nextReadIndex) + curWriteIndex;
        }
      }
      @Override
      public Entry<KeyType> next()
      {
        if (!hasNext()) {
          throw new NoSuchElementException();
        }
         Here, nextReadIndex should be valid which means:
         - a valid array index which should be >= 0 and < maxNumSlots
         - an index of the array slot where the aggregation for the corresponding grouping key is done
         - an index of the array slot which is not read yet
        final int recordOffset = recordSize * nextReadIndex;
        keySerde.readFromByteBuffer(reusableEntry.getKey(), buffer, recordOffset);
        for (int i = 0; i < aggregators.length; i++) {
          reusableEntry.getValues()[i] = aggregators[i].get(buffer, recordOffset + aggregatorOffsets[i]);
        }
        final int targetIndex = nextReadIndex == maxNumSlots - 1 ? 0 : nextReadIndex + 1;
         Wait if the array is empty until at least one slot becomes available for read, and then increase
         nextReadIndex.
        increaseReadIndexTo(targetIndex);
        return reusableEntry;
      }
      /**
       * Wait for {@link StreamingMergeSortedGrouper#curWriteIndex} to be moved if necessary and move
       * {@link StreamingMergeSortedGrouper#nextReadIndex}.
       *
       * @param target the target index {@link StreamingMergeSortedGrouper#nextReadIndex} will move to
       */
      private void increaseReadIndexTo(int target)
      {
         Check that the array is empty and wait for at least one slot to become available.
         curWriteIndex is a volatile variable and the changes on it are continuously checked until they are seen in
         the while loop. See the following links.
         * http:docs.oracle.com/javase/specs/jls/se7/html/jls-8.html#jls-8.3.1.4
         * http:docs.oracle.com/javase/specs/jls/se7/html/jls-17.html#jls-17.4.5
         * https:stackoverflow.com/questions/11761552/detailed-semantics-of-volatile-regarding-timeliness-of-visibility
        final long startAtNs = System.nanoTime();
        final long queryTimeoutAtNs = getQueryTimeoutAtNs(startAtNs);
        final long spinTimeoutAtNs = startAtNs + SPIN_FOR_TIMEOUT_THRESHOLD_NS;
        long timeoutNs = queryTimeoutAtNs - startAtNs;
        long spinTimeoutNs = SPIN_FOR_TIMEOUT_THRESHOLD_NS;
         The below condition is checked in a while loop instead of using a lock to avoid frequent thread park.
        while ((curWriteIndex == -1 || target == curWriteIndex) &&
               !finished && !Thread.currentThread().isInterrupted()) {
          if (timeoutNs <= 0L) {
            throw new QueryTimeoutException();
          }
           Thread.yield() should not be called from the very beginning
          if (spinTimeoutNs <= 0L) {
            Thread.yield();
          }
          long now = System.nanoTime();
          timeoutNs = queryTimeoutAtNs - now;
          spinTimeoutNs = spinTimeoutAtNs - now;
        }
         Changes on curWriteIndex happens-before changing nextReadIndex.
        nextReadIndex = target;
      }
      @Override
      public void close()
      {
         do nothing
      }
    };
  }
public CloseableIterator < Entry < KeyType > > iterator ( ) { if ( ! initialized ) { throw new ISE ( "Grouper should be initialized first" ) ; } return new CloseableIterator < Entry < KeyType > > ( ) { final ReusableEntry < KeyType > reusableEntry = ReusableEntry . create ( keySerde , aggregators . length ) ; { increaseReadIndexTo ( 0 ) ; } @ Override public boolean hasNext ( ) { return ! finished || remaining ( ) > 0 ; } /**
       * Calculate the number of remaining items in the array.  Must be called only when
       * {@link StreamingMergeSortedGrouper#finished} is true.
       *
       * @return the number of remaining items
       */ private int remaining ( ) { if ( curWriteIndex >= nextReadIndex ) { return curWriteIndex - nextReadIndex ; } else { return ( maxNumSlots - nextReadIndex ) + curWriteIndex ; } } @ Override public Entry < KeyType > next ( ) { if ( ! hasNext ( ) ) { throw new NoSuchElementException ( ) ; } final int recordOffset = recordSize * nextReadIndex ; keySerde . readFromByteBuffer ( reusableEntry . getKey ( ) , buffer , recordOffset ) ; for ( int i = 0 ; i < aggregators . length ; i ++ ) { reusableEntry . getValues ( ) [ i ] = aggregators [ i ] . get ( buffer , recordOffset + aggregatorOffsets [ i ] ) ; } final int targetIndex = nextReadIndex == maxNumSlots - 1 ? 0 : nextReadIndex + 1 ; increaseReadIndexTo ( targetIndex ) ; return reusableEntry ; } /**
       * Wait for {@link StreamingMergeSortedGrouper#curWriteIndex} to be moved if necessary and move
       * {@link StreamingMergeSortedGrouper#nextReadIndex}.
       *
       * @param target the target index {@link StreamingMergeSortedGrouper#nextReadIndex} will move to
       */ private void increaseReadIndexTo ( int target ) { final long startAtNs = System . nanoTime ( ) ; final long queryTimeoutAtNs = getQueryTimeoutAtNs ( startAtNs ) ; final long spinTimeoutAtNs = startAtNs + SPIN_FOR_TIMEOUT_THRESHOLD_NS ; long timeoutNs = queryTimeoutAtNs - startAtNs ; long spinTimeoutNs = SPIN_FOR_TIMEOUT_THRESHOLD_NS ; while ( ( curWriteIndex == - 1 || target == curWriteIndex ) && ! finished && ! Thread . currentThread ( ) . isInterrupted ( ) ) { if ( timeoutNs <= 0L ) { throw new QueryTimeoutException ( ) ; } if ( spinTimeoutNs <= 0L ) { Thread . yield ( ) ; } long now = System . nanoTime ( ) ; timeoutNs = queryTimeoutAtNs - now ; spinTimeoutNs = spinTimeoutAtNs - now ; } nextReadIndex = target ; } @ Override public void close ( ) { } } ; }
Return a sorted iterator.
Return a sorted iterator .
// Wait for some data to be ready and initialize nextReadIndex. // If setting finished happens-before the below check, curWriteIndex isn't changed anymore and thus remainig() // can be computed safely because nextReadIndex is changed only by the reading thread. // Otherwise, hasNext() always returns true. // // The below line can be executed between increasing curWriteIndex and setting finished in // StreamingMergeSortedGrouper.finish(), but it is also a valid case because there should be at least one slot // which is not read yet before finished is set. // Here, nextReadIndex should be valid which means: // - a valid array index which should be >= 0 and < maxNumSlots // - an index of the array slot where the aggregation for the corresponding grouping key is done // - an index of the array slot which is not read yet // Wait if the array is empty until at least one slot becomes available for read, and then increase // nextReadIndex. // Check that the array is empty and wait for at least one slot to become available. // // curWriteIndex is a volatile variable and the changes on it are continuously checked until they are seen in // the while loop. See the following links. // * http://docs.oracle.com/javase/specs/jls/se7/html/jls-8.html#jls-8.3.1.4 // * http://docs.oracle.com/javase/specs/jls/se7/html/jls-17.html#jls-17.4.5 // * https://stackoverflow.com/questions/11761552/detailed-semantics-of-volatile-regarding-timeliness-of-visibility // The below condition is checked in a while loop instead of using a lock to avoid frequent thread park. // Thread.yield() should not be called from the very beginning // Changes on curWriteIndex happens-before changing nextReadIndex. // do nothing