How to merge iterator parser

Question

I have an Iterator[(A1,B1)] and two functions

• fA: (Iterator[A1]) => Iterator[A2] and
• fB: (Iterator[B1]) => Iterator[B2].

Is it possible to make a fAB: (Iterator[(A1,B1)]) => Iterator[(A2,B2)] without converting Iterators to Seq?

Edit

Both answers below are good. I selected @Aivean's answer because the code is simpler and it uses specialized scala data structure (Stream).

The only drawback is the stackoverfow limitation but it shouldn't be a problem for most use cases. If your iterator can be very (very) long, then @Alexey's solution should be preferred.

Answer 1

No. Your hypothetical function has to call one of fA and fB first. Let's say it calls fA and it requests all the A1s before producing anything. Then you don't have any B1s remaining to pass to fB, unless you save them somewhere, potentially leaking memory. If that's acceptable, you can do:

def unzip[A, B](iter: Iterator[(A, B)]) = {
  var qA = Queue.empty[A]
  var qB = Queue.empty[B]

  val iterA = new Iterator[A] {
    override def hasNext = qA.nonEmpty || iter.hasNext

    override def next() = qA.dequeueOption match {
      case Some((a, qA1)) =>
        qA = qA1
        a
      case None =>
        val (a, b) = iter.next()
        qB = qB.enqueue(b)
        a
    }
  }

  // similar iterB

  (iterA, iterB)
}

and then

val (iterA, iterB) = unzip(iterator)
fA(iterAfA).zip(fB(iterB))

(Well, you can also write iterator => fA(iterator.map(_._1)).zip(fB(iterator.map(_._2)): it has the right type, but is probably not what you want. Namely, it will use only one field of each tuple produced by the original iterator, and drop the other.)

Answer 2

I came to much simpler implementation:

def iterUnzip[A1, B1, A2, B2](it: Iterator[(A1, B1)],
                           fA: (Iterator[A1]) => Iterator[A2],
                           fB: (Iterator[B1]) => Iterator[B2]) =
  it.toStream match {
    case s => fA(s.map(_._1).toIterator).zip(fB(s.map(_._2).toIterator))
  }

The idea is to convert iterator to stream. Stream in Scala is lazy but also provides memoization. This effectively provides the same buffering mechanism, as in @AlexeyRomanov's solution, but more concise. The only drawback is that Stream stores memoized elements on stack as opposed to the explicit Queue, thus if fA and fB produce elements on uneven rate, you may get StackOverflow exception.

Test that evaluation is lazy indeed:

val iter = Stream.from(0).map(x => (x, x + 1))
  .map(x => {println("fetched: " + x); x}).take(5).toIterator

iterUnzip(
  iter,
  (_:Iterator[Int]).flatMap(x => List(x, x)),
  (_:Iterator[Int]).map(_ + 1)
).toList

Result:

fetched: (0,1)
iter: Iterator[(Int, Int)] = non-empty iterator

fetched: (1,2)
fetched: (2,3)
fetched: (3,4)
fetched: (4,5)
res0: List[(Int, Int)] = List((0,2), (0,3), (1,4), (1,5), (2,6))

I also tried reasonably hard to get StackOverflow exception by producing uneven iterators, but failed.

val iter = Stream.from(0).map(x => (x, x + 1)).take(10000000).toIterator
iterUnzip(
    iter,
    (_:Iterator[Int]).flatMap(x => List.fill(1000000)(x)),
    (_:Iterator[Int]).map(_ + 1)
  ).size

Works fine on -Xss5m and produces:

res10: Int = 10000000

So, overall this is reasonably good and concise solution, unless you have some extreme usecases.