ReusingBlockResettableIterator extends NonReusingBlockResettableIterator<T> { @Override public boolean hasNext() { try { if (this.nextElement == null) { if (this.readPhase) { T tmp = getNextRecord(this.reuseElement); if (tmp != null) { this.nextElement = tmp; return true; } else { return false; } } else { if (this.input.hasNext()) { final T next = this.input.next(); if (writeNextRecord(next)) { this.nextElement = next; return true; } else { this.leftOverElement = next; return false; } } else { this.noMoreBlocks = true; return false; } } } else { return true; } } catch (IOException ioex) { throw new RuntimeException("Error (de)serializing record in block resettable iterator.", ioex); } } ReusingBlockResettableIterator(MemoryManager memoryManager, Iterator<T> input,
			TypeSerializer<T> serializer, int numPages,
			AbstractInvokable ownerTask);  ReusingBlockResettableIterator(MemoryManager memoryManager, TypeSerializer<T>
			serializer, int numPages, AbstractInvokable ownerTask); @Override boolean hasNext(); static final Logger LOG; }
@Test public void testSerialBlockResettableIterator() throws Exception { final AbstractInvokable memOwner = new DummyInvokable(); final ReusingBlockResettableIterator<Record> iterator = new ReusingBlockResettableIterator<Record>( this.memman, this.reader, this.serializer, 1, memOwner); iterator.open(); int lower = 0; int upper = 0; do { lower = upper; upper = lower; while (iterator.hasNext()) { Record target = iterator.next(); int val = target.getField(0, IntValue.class).getValue(); Assert.assertEquals(upper++, val); } for (int i = 0; i < 5; ++i) { iterator.reset(); int count = 0; while (iterator.hasNext()) { Record target = iterator.next(); int val = target.getField(0, IntValue.class).getValue(); Assert.assertEquals(lower + (count++), val); } Assert.assertEquals(upper - lower, count); } } while (iterator.nextBlock()); Assert.assertEquals(NUM_VALUES, upper); iterator.close(); }
@Test public void testDoubleBufferedBlockResettableIterator() throws Exception { final AbstractInvokable memOwner = new DummyInvokable(); final ReusingBlockResettableIterator<Record> iterator = new ReusingBlockResettableIterator<Record>( this.memman, this.reader, this.serializer, 2, memOwner); iterator.open(); int lower = 0; int upper = 0; do { lower = upper; upper = lower; while (iterator.hasNext()) { Record target = iterator.next(); int val = target.getField(0, IntValue.class).getValue(); Assert.assertEquals(upper++, val); } for (int i = 0; i < 5; ++i) { iterator.reset(); int count = 0; while (iterator.hasNext()) { Record target = iterator.next(); int val = target.getField(0, IntValue.class).getValue(); Assert.assertEquals(lower + (count++), val); } Assert.assertEquals(upper - lower, count); } } while (iterator.nextBlock()); Assert.assertEquals(NUM_VALUES, upper); iterator.close(); }
@Test public void testTwelveFoldBufferedBlockResettableIterator() throws Exception { final AbstractInvokable memOwner = new DummyInvokable(); final ReusingBlockResettableIterator<Record> iterator = new ReusingBlockResettableIterator<Record>( this.memman, this.reader, this.serializer, 12, memOwner); iterator.open(); int lower = 0; int upper = 0; do { lower = upper; upper = lower; while (iterator.hasNext()) { Record target = iterator.next(); int val = target.getField(0, IntValue.class).getValue(); Assert.assertEquals(upper++, val); } for (int i = 0; i < 5; ++i) { iterator.reset(); int count = 0; while (iterator.hasNext()) { Record target = iterator.next(); int val = target.getField(0, IntValue.class).getValue(); Assert.assertEquals(lower + (count++), val); } Assert.assertEquals(upper - lower, count); } } while (iterator.nextBlock()); Assert.assertEquals(NUM_VALUES, upper); iterator.close(); }