GroupingShuffleRangeTracker implements RangeTracker<ShufflePosition> { @Override public synchronized boolean tryReturnRecordAt( boolean isAtSplitPoint, @Nullable ShufflePosition groupStart) { if (lastGroupStart == null && !isAtSplitPoint) { throw new IllegalStateException( String.format("The first group [at %s] must be at a split point", groupStart.toString())); } if (this.startPosition != null && groupStart.compareTo(this.startPosition) < 0) { throw new IllegalStateException( String.format( "Trying to return record at %s which is before the starting position at %s", groupStart, this.startPosition)); } int comparedToLast = (lastGroupStart == null) ? 1 : groupStart.compareTo(this.lastGroupStart); if (comparedToLast < 0) { throw new IllegalStateException( String.format( "Trying to return group at %s which is before the last-returned group at %s", groupStart, this.lastGroupStart)); } if (isAtSplitPoint) { splitPointsSeen++; if (comparedToLast == 0) { throw new IllegalStateException( String.format( "Trying to return a group at a split point with same position as the " + "previous group: both at %s, last group was %s", groupStart, lastGroupWasAtSplitPoint ? "at a split point." : "not at a split point.")); } if (stopPosition != null && groupStart.compareTo(stopPosition) >= 0) { return false; } } else { checkState( comparedToLast == 0, "Trying to return a group not at a split point, but with a different position " + "than the previous group: last group was %s at %s, current at %s", lastGroupWasAtSplitPoint ? "a split point" : "a non-split point", lastGroupStart, groupStart); } this.lastGroupStart = groupStart; this.lastGroupWasAtSplitPoint = isAtSplitPoint; return true; } GroupingShuffleRangeTracker(
      @Nullable ShufflePosition startPosition, @Nullable ShufflePosition stopPosition); @Override ShufflePosition getStartPosition(); @Override synchronized ShufflePosition getStopPosition(); synchronized ShufflePosition getLastGroupStart(); synchronized boolean isStarted(); synchronized boolean isDone(); synchronized long getSplitPointsProcessed(); @Override synchronized boolean tryReturnRecordAt(
      boolean isAtSplitPoint, @Nullable ShufflePosition groupStart); @Override synchronized boolean trySplitAtPosition(ShufflePosition splitPosition); @Override synchronized double getFractionConsumed(); @Override synchronized String toString();  }
@Test public void testTryReturnRecordInfiniteRange() throws Exception { GroupingShuffleRangeTracker tracker = new GroupingShuffleRangeTracker(null, null); assertTrue(tracker.tryReturnRecordAt(true, ofBytes(1, 2, 3))); assertTrue(tracker.tryReturnRecordAt(true, ofBytes(1, 2, 5))); assertTrue(tracker.tryReturnRecordAt(true, ofBytes(3, 6, 8, 10))); }
@Test public void testTryReturnRecordWithNonSplitPoints() throws Exception { GroupingShuffleRangeTracker tracker = new GroupingShuffleRangeTracker(ofBytes(1, 0, 0), ofBytes(5, 0, 0)); assertTrue(tracker.tryReturnRecordAt(true, ofBytes(1, 2, 3))); assertTrue(tracker.tryReturnRecordAt(false, ofBytes(1, 2, 3))); assertTrue(tracker.tryReturnRecordAt(false, ofBytes(1, 2, 3))); assertTrue(tracker.tryReturnRecordAt(true, ofBytes(1, 2, 5))); assertTrue(tracker.tryReturnRecordAt(false, ofBytes(1, 2, 5))); assertTrue(tracker.tryReturnRecordAt(true, ofBytes(3, 6, 8, 10))); assertTrue(tracker.tryReturnRecordAt(true, ofBytes(4, 255, 255, 255, 255))); }
@Test public void testFirstRecordNonSplitPoint() throws Exception { GroupingShuffleRangeTracker tracker = new GroupingShuffleRangeTracker(ofBytes(3, 0, 0), ofBytes(5, 0, 0)); expected.expect(IllegalStateException.class); tracker.tryReturnRecordAt(false, ofBytes(3, 4, 5)); }
@Test public void testNonSplitPointRecordWithDifferentPosition() throws Exception { GroupingShuffleRangeTracker tracker = new GroupingShuffleRangeTracker(ofBytes(3, 0, 0), ofBytes(5, 0, 0)); tracker.tryReturnRecordAt(true, ofBytes(3, 4, 5)); expected.expect(IllegalStateException.class); tracker.tryReturnRecordAt(false, ofBytes(3, 4, 6)); }
@Test public void testTryReturnRecordBeforeStart() throws Exception { GroupingShuffleRangeTracker tracker = new GroupingShuffleRangeTracker(ofBytes(3, 0, 0), ofBytes(5, 0, 0)); expected.expect(IllegalStateException.class); tracker.tryReturnRecordAt(true, ofBytes(1, 2, 3)); }
@Test public void testTryReturnNonMonotonic() throws Exception { GroupingShuffleRangeTracker tracker = new GroupingShuffleRangeTracker(ofBytes(3, 0, 0), ofBytes(5, 0, 0)); tracker.tryReturnRecordAt(true, ofBytes(3, 4, 5)); tracker.tryReturnRecordAt(true, ofBytes(3, 4, 6)); expected.expect(IllegalStateException.class); tracker.tryReturnRecordAt(true, ofBytes(3, 2, 1)); }
@Test public void testTryReturnIdenticalPositions() throws Exception { GroupingShuffleRangeTracker tracker = new GroupingShuffleRangeTracker(ofBytes(3, 0, 0), ofBytes(5, 0, 0)); tracker.tryReturnRecordAt(true, ofBytes(3, 4, 5)); expected.expect(IllegalStateException.class); tracker.tryReturnRecordAt(true, ofBytes(3, 4, 5)); }