@Test public void emitCartesianProduct() { final Iterator<Tuple2<BatchJoinKey<String>, Either<String, String>>> inner = asList( entry("key1", LEFT, Either.left("v1")), entry("key1", LEFT, Either.left("v2")), entry("key1", RIGHT, Either.right("w1")), entry("key1", RIGHT, Either.right("w2")), entry("key1", RIGHT, Either.right("w3"))) .iterator(); final List<Tuple2<String, Tuple2<Optional<String>, Optional<String>>>> results = Lists.newArrayList(new BatchJoinIterator<>(inner)); assertEquals( asList( entry("key1", "v1", "w1"), entry("key1", "v2", "w1"), entry("key1", "v1", "w2"), entry("key1", "v2", "w2"), entry("key1", "v1", "w3"), entry("key1", "v2", "w3")), results); }
private void emitCartesianProduct(Tuple2<BatchJoinKey<K>, Either<L, R>> right) { leftQueue.forEach( left -> outQueue.add( new Tuple2<>( left._1.getKey(), new Tuple2<>(Optional.of(left._2.left()), Optional.of(right._2.right()))))); }
BatchJoinIterator implements Iterator<Tuple2<K, Tuple2<Optional<L>, Optional<R>>>> { private void emitCartesianProduct(Tuple2<BatchJoinKey<K>, Either<L, R>> right) { leftQueue.forEach( left -> outQueue.add( new Tuple2<>( left._1.getKey(), new Tuple2<>(Optional.of(left._2.left()), Optional.of(right._2.right()))))); } }
BatchJoinIterator implements Iterator<Tuple2<K, Tuple2<Optional<L>, Optional<R>>>> { private void emitCartesianProduct(Tuple2<BatchJoinKey<K>, Either<L, R>> right) { leftQueue.forEach( left -> outQueue.add( new Tuple2<>( left._1.getKey(), new Tuple2<>(Optional.of(left._2.left()), Optional.of(right._2.right()))))); } BatchJoinIterator(Iterator<Tuple2<BatchJoinKey<K>, Either<L, R>>> inner); }
BatchJoinIterator implements Iterator<Tuple2<K, Tuple2<Optional<L>, Optional<R>>>> { private void emitCartesianProduct(Tuple2<BatchJoinKey<K>, Either<L, R>> right) { leftQueue.forEach( left -> outQueue.add( new Tuple2<>( left._1.getKey(), new Tuple2<>(Optional.of(left._2.left()), Optional.of(right._2.right()))))); } BatchJoinIterator(Iterator<Tuple2<BatchJoinKey<K>, Either<L, R>>> inner); @Override boolean hasNext(); @Override Tuple2<K, Tuple2<Optional<L>, Optional<R>>> next(); }
BatchJoinIterator implements Iterator<Tuple2<K, Tuple2<Optional<L>, Optional<R>>>> { private void emitCartesianProduct(Tuple2<BatchJoinKey<K>, Either<L, R>> right) { leftQueue.forEach( left -> outQueue.add( new Tuple2<>( left._1.getKey(), new Tuple2<>(Optional.of(left._2.left()), Optional.of(right._2.right()))))); } BatchJoinIterator(Iterator<Tuple2<BatchJoinKey<K>, Either<L, R>>> inner); @Override boolean hasNext(); @Override Tuple2<K, Tuple2<Optional<L>, Optional<R>>> next();  }