MergeIterator implements MutableObjectIterator<E> { @Override public boolean next(E target) throws IOException { if (this.heap.size() > 0) { final HeadStream<E> top = this.heap.peek(); this.serializer.copyTo(top.getHead(), target); if (!top.nextHead()) { this.heap.poll(); } else { this.heap.adjustTop(); } return true; } else { return false; } } MergeIterator(List<MutableObjectIterator<E>> iterators,
			TypeSerializer<E> serializer, TypeComparator<E> comparator); @Override boolean next(E target);  }
@Test public void testMergeOfTwoStreams() throws Exception { List<MutableObjectIterator<PactRecord>> iterators = new ArrayList<MutableObjectIterator<PactRecord>>(); 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}; Comparator<TestData.Key> comparator = new TestData.KeyComparator(); MutableObjectIterator<PactRecord> iterator = new MergeIterator<PactRecord>(iterators, this.serializer, this.comparator); PactRecord rec1 = new PactRecord(); PactRecord rec2 = new PactRecord(); final Key k1 = new Key(); final Key k2 = new Key(); int pos = 1; Assert.assertTrue(iterator.next(rec1)); Assert.assertEquals(expected[0], rec1.getField(0, TestData.Key.class).getKey()); while (iterator.next(rec2)) { k1.setKey(rec1.getField(0, TestData.Key.class).getKey()); k2.setKey(rec2.getField(0, TestData.Key.class).getKey()); Assert.assertTrue(comparator.compare(k1, k2) <= 0); Assert.assertEquals(expected[pos++], k2.getKey()); PactRecord tmp = rec1; rec1 = rec2; rec2 = tmp; } }
@Test public void testMergeOfTenStreams() throws Exception { List<MutableObjectIterator<PactRecord>> iterators = new ArrayList<MutableObjectIterator<PactRecord>>(); 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" })); Comparator<TestData.Key> comparator = new TestData.KeyComparator(); MutableObjectIterator<PactRecord> iterator = new MergeIterator<PactRecord>(iterators, this.serializer, this.comparator); int elementsFound = 1; PactRecord rec1 = new PactRecord(); PactRecord rec2 = new PactRecord(); final Key k1 = new Key(); final Key k2 = new Key(); Assert.assertTrue(iterator.next(rec1)); while (iterator.next(rec2)) { elementsFound++; k1.setKey(rec1.getField(0, TestData.Key.class).getKey()); k2.setKey(rec2.getField(0, TestData.Key.class).getKey()); Assert.assertTrue(comparator.compare(k1, k2) <= 0); PactRecord tmp = rec1; rec1 = rec2; rec2 = tmp; } Assert.assertEquals("Too few elements returned from stream.", 50, elementsFound); }
@Test public void testInvalidMerge() throws Exception { List<MutableObjectIterator<PactRecord>> iterators = new ArrayList<MutableObjectIterator<PactRecord>>(); 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" })); Comparator<TestData.Key> comparator = new TestData.KeyComparator(); MutableObjectIterator<PactRecord> iterator = new MergeIterator<PactRecord>(iterators, this.serializer, this.comparator); boolean violationFound = false; PactRecord rec1 = new PactRecord(); PactRecord rec2 = new PactRecord(); Assert.assertTrue(iterator.next(rec1)); while (iterator.next(rec2)) { final Key k1 = new Key(); final Key k2 = new Key(); k1.setKey(rec1.getField(0, TestData.Key.class).getKey()); k2.setKey(rec2.getField(0, TestData.Key.class).getKey()); if (comparator.compare(k1, k2) > 0) { violationFound = true; break; } PactRecord tmp = rec1; rec1 = rec2; rec2 = tmp; } Assert.assertTrue("Merge must have returned a wrong result", violationFound); }