MergeIterator implements MutableObjectIterator<E> { @Override public E next(E reuse) throws IOException { if (this.heap.size() > 0) { final HeadStream<E> top = this.heap.peek(); E result = top.getHead(); if (!top.nextHead(reuse)) { this.heap.poll(); } else { this.heap.adjustTop(); } return result; } else { return null; } } MergeIterator(List<MutableObjectIterator<E>> iterators, TypeComparator<E> comparator); @Override E next(E reuse); @Override E next();  }
@Test public void testMergeOfTwoStreams() throws Exception { List<MutableObjectIterator<Tuple2<Integer, String>>> iterators = new ArrayList<>(); iterators.add(newIterator(new int[] { 1, 2, 4, 5, 10 }, new String[] { "1", "2", "4", "5", "10" })); iterators.add(newIterator(new int[] { 3, 6, 7, 10, 12 }, new String[] { "3", "6", "7", "10", "12" })); final int[] expected = new int[] {1, 2, 3, 4, 5, 6, 7, 10, 10, 12}; TypeComparator<Integer> comparator = new IntComparator(true); MutableObjectIterator<Tuple2<Integer, String>> iterator = new MergeIterator<>(iterators, this.comparator); Tuple2<Integer, String> rec1 = new Tuple2<>(); Tuple2<Integer, String> rec2 = new Tuple2<>(); int k1 = 0; int k2 = 0; int pos = 1; Assert.assertTrue((rec1 = iterator.next(rec1)) != null); Assert.assertEquals(expected[0], rec1.f0.intValue()); while ((rec2 = iterator.next(rec2)) != null) { k1 = rec1.f0; k2 = rec2.f0; Assert.assertTrue(comparator.compare(k1, k2) <= 0); Assert.assertEquals(expected[pos++], k2); Tuple2<Integer, String> tmp = rec1; rec1 = rec2; rec2 = tmp; } }
@Test public void testMergeOfTenStreams() throws Exception { List<MutableObjectIterator<Tuple2<Integer, String>>> iterators = new ArrayList<>(); iterators.add(newIterator(new int[] { 1, 2, 17, 23, 23 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 2, 6, 7, 8, 9 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 4, 10, 11, 11, 12 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 3, 6, 7, 10, 12 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 7, 10, 15, 19, 44 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 6, 6, 11, 17, 18 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 1, 2, 4, 5, 10 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 5, 10, 19, 23, 29 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 9, 9, 9, 9, 9 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 8, 8, 14, 14, 15 }, new String[] { "A", "B", "C", "D", "E" })); TypeComparator<Integer> comparator = new IntComparator(true); MutableObjectIterator<Tuple2<Integer, String>> iterator = new MergeIterator<>(iterators, this.comparator); int elementsFound = 1; Tuple2<Integer, String> rec1 = new Tuple2<>(); Tuple2<Integer, String> rec2 = new Tuple2<>(); Assert.assertTrue((rec1 = iterator.next(rec1)) != null); while ((rec2 = iterator.next(rec2)) != null) { elementsFound++; Assert.assertTrue(comparator.compare(rec1.f0, rec2.f0) <= 0); Tuple2<Integer, String> tmp = rec1; rec1 = rec2; rec2 = tmp; } Assert.assertEquals("Too few elements returned from stream.", 50, elementsFound); }
@Test public void testInvalidMerge() throws Exception { List<MutableObjectIterator<Tuple2<Integer, String>>> iterators = new ArrayList<>(); iterators.add(newIterator(new int[] { 1, 2, 17, 23, 23 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 2, 6, 7, 8, 9 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 4, 10, 11, 11, 12 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 3, 6, 10, 7, 12 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 7, 10, 15, 19, 44 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 6, 6, 11, 17, 18 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 1, 2, 4, 5, 10 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 5, 10, 19, 23, 29 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 9, 9, 9, 9, 9 }, new String[] { "A", "B", "C", "D", "E" })); iterators.add(newIterator(new int[] { 8, 8, 14, 14, 15 }, new String[] { "A", "B", "C", "D", "E" })); TypeComparator<Integer> comparator = new IntComparator(true); MutableObjectIterator<Tuple2<Integer, String>> iterator = new MergeIterator<>(iterators, this.comparator); boolean violationFound = false; Tuple2<Integer, String> rec1 = new Tuple2<>(); Tuple2<Integer, String> rec2 = new Tuple2<>(); Assert.assertTrue((rec1 = iterator.next(rec1)) != null); while ((rec2 = iterator.next(rec2)) != null) { if (comparator.compare(rec1.f0, rec2.f0) > 0) { violationFound = true; break; } Tuple2<Integer, String> tmp = rec1; rec1 = rec2; rec2 = tmp; } Assert.assertTrue("Merge must have returned a wrong result", violationFound); }