InMemoryReader extends NativeReader<T> { @Override public InMemoryReaderIterator iterator() throws IOException { return new InMemoryReaderIterator(); } InMemoryReader(
      List<String> encodedElements,
      @Nullable Integer startIndex,
      @Nullable Integer endIndex,
      Coder<T> coder); @Override InMemoryReaderIterator iterator();  }
@Test public void testProgressReporting() throws Exception { List<Integer> elements = Arrays.asList(33, 44, 55, 66, 77, 88); Coder<Integer> coder = BigEndianIntegerCoder.of(); InMemoryReader<Integer> inMemoryReader = new InMemoryReader<>(encodedElements(elements, coder), 1, 4, coder); try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) { assertNull(iterator.getProgress()); assertEquals(3, iterator.getRemainingParallelism(), 0.0); assertTrue(iterator.start()); Assert.assertEquals( ReaderTestUtils.positionAtIndex(1L), ReaderTestUtils.positionFromProgress(iterator.getProgress())); assertEquals(3, iterator.getRemainingParallelism(), 0.0); assertTrue(iterator.advance()); Assert.assertEquals( ReaderTestUtils.positionAtIndex(2L), ReaderTestUtils.positionFromProgress(iterator.getProgress())); assertEquals(2, iterator.getRemainingParallelism(), 0.0); assertTrue(iterator.advance()); Assert.assertEquals( ReaderTestUtils.positionAtIndex(3L), ReaderTestUtils.positionFromProgress(iterator.getProgress())); assertEquals(1, iterator.getRemainingParallelism(), 0.0); assertFalse(iterator.advance()); } }
@Test public void testDynamicSplit() throws Exception { List<Integer> elements = Arrays.asList(33, 44, 55, 66, 77, 88); Coder<Integer> coder = BigEndianIntegerCoder.of(); InMemoryReader<Integer> inMemoryReader = new InMemoryReader<>(encodedElements(elements, coder), 1, 4, coder); try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) { assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(3L))); } try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) { assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(3L))); assertTrue(iterator.start()); assertNull( iterator.requestDynamicSplit(toDynamicSplitRequest(new ApproximateSplitRequest()))); assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(null))); } try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) { assertTrue(iterator.start()); NativeReader.DynamicSplitResult dynamicSplitResult = iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(3L)); Assert.assertEquals( ReaderTestUtils.positionAtIndex(3L), ReaderTestUtils.positionFromSplitResult(dynamicSplitResult)); assertEquals(3, iterator.tracker.getStopPosition().longValue()); assertEquals(44, iterator.getCurrent().intValue()); assertTrue(iterator.advance()); assertEquals(55, iterator.getCurrent().intValue()); assertFalse(iterator.advance()); } try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) { assertTrue(iterator.start()); assertEquals(44, iterator.getCurrent().intValue()); assertTrue(iterator.advance()); assertEquals(55, iterator.getCurrent().intValue()); assertTrue(iterator.advance()); assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(3L))); assertEquals(4, iterator.tracker.getStopPosition().longValue()); assertEquals(66, iterator.getCurrent().intValue()); assertFalse(iterator.advance()); } try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) { assertTrue(iterator.start()); assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(5L))); assertEquals(4, iterator.tracker.getStopPosition().longValue()); } }
@Test public void testParallelism() throws Exception { List<Integer> elements = Arrays.asList(33, 44, 55, 66, 77, 88); Coder<Integer> coder = BigEndianIntegerCoder.of(); InMemoryReader<Integer> inMemoryReader = new InMemoryReader<>(encodedElements(elements, coder), 1, 4, coder); int count = 0; try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) { for (boolean more = iterator.start(); more; more = iterator.advance()) { assertTrue(iterator.getRemainingParallelism() >= 1); assertEquals(3 - count, iterator.getRemainingParallelism(), 0 ); count++; } } }