Java stream's iterator forces flatmap to traverse substream before getting the first item

user_2551580 0
Viewed 346

I have the need to create an iterator out of a stream of streams. Both the parent and the child streams are composed by non-interfering stateless operations and the obvious strategy is to use flatMap.

Turns out that iterator, at the first "hasNext" invocation, traverse the entire first substream and I don't understand why. Despite iterator() is a terminal operation is clearly stated that it shouldn't consume the stream. I need that the objects generated from the substream are generated one by one.

To replicate the behaviour I've mocked my real code with a sample which shows the same:

import java.util.Iterator;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

public class FreeRunner {

    public static void main(String[] args) {
        AtomicInteger x = new AtomicInteger();
        Iterator<C> iterator = Stream.generate(() -> null)
                .takeWhile(y -> x.incrementAndGet() < 5)
                .filter(y -> x.get() % 2 == 0)
                .map(n -> new A("A" + x.get()))
                .flatMap(A::getBStream)
                .filter(Objects::nonNull)
                .map(B::toC)
                .iterator();

        while(iterator.hasNext()) {
            System.out.println("after hasNext()");
            C next = iterator.next();
            System.out.println(next);
        }

    }

    private static class A {
        private final String name;

        public A(String name) {
            this.name = name;
            System.out.println(" > created " + name);
        }

        public Stream<B> getBStream() {
            AtomicInteger c = new AtomicInteger();
            return Stream.generate(() -> null)
                    .takeWhile(x -> c.incrementAndGet() < 5)
                    .map(n -> c.get() % 2 == 0 ? null : new B(this.name + "->B" + c.get()));
        }

        public String toString() {
            return name;
        }
    }

    private static class B {

        private final String name;

        public B(String name) {
            this.name = name;
            System.out.println(" >> created " + name);
        }

        public String toString() {
            return name;
        }

        public C toC() {
            return new C(this.name + "+C");
        }

    }

    private static class C {

        private final String name;

        public C(String name) {
            this.name = name;
            System.out.println(" >>> created " + name);
        }

        public String toString() {
            return name;
        }
    }
}

When it is executed it shows:

 > created A2
 >> created A2->B1
 >>> created A2->B1+C
 >> created A2->B3
 >>> created A2->B3+C
after hasNext()
A2->B1+C
after hasNext()
A2->B3+C
 > created A4
 >> created A4->B1
 >>> created A4->B1+C
 >> created A4->B3
 >>> created A4->B3+C
after hasNext()
A4->B1+C
after hasNext()
A4->B3+C

Process finished with exit code 0

In debug it's clear that iterator.hasNext() triggers the generation of objects B and C.

The desired behaviour, instead, is:

 > created A2
 >> created A2->B1
 >>> created A2->B1+C
after hasNext()
A2->B1+C
 >> created A2->B3
 >>> created A2->B3+C
after hasNext()
A2->B3+C
 > created A4
 >> created A4->B1
 >>> created A4->B1+C
after hasNext()
A4->B1+C
 >> created A4->B3
 >>> created A4->B3+C
after hasNext()
A4->B3+C

What am I missing here?

java-stream iterator java
3 Answers

I found a way out, but I had to sacrifice the laziness of the primary stream. As I posted in the comment above the problem which I tried to simplify mocking the code is about to read an excel file sheet by sheet (filtered by sheet's name) and traversing all rows to create objects accordingly to the data in the spreadsheet.

The original idea is still good to me but, apparently, Stream.iterator() implementation consumes each nested stream at the first hasNext() invocation operated at the creation of the first A object.

So I abandoned flatMap() and used reduce(Stream::concat) to concatenate all streams generated by A.getBStream():

    public static void main(String[] args) {
        AtomicInteger x = new AtomicInteger();
        Iterator<C> it = Stream.generate(() -> null)
                .takeWhile(y -> x.incrementAndGet() < 5)
                .filter(y -> x.get() % 2 == 0)
                .map(a -> new A("A" + x.get()))
                .map(A::getBStream)
                .filter(Objects::nonNull)
                .reduce(Stream::concat)
                .orElseGet(Stream::empty)
                .filter(Objects::nonNull)
                .map(B::toC)
                .iterator();

        while(it.hasNext()) {
            System.out.println("after hasNext()");
            C next = it.next();
            System.out.println(next);
        }
    }

This produces the following output:

 > created A2
 > created A4
 >> created A2->B0
 >>> created A2->B0+C
after hasNext()
A2->B0+C
 >> created A2->B1
 >>> created A2->B1+C
after hasNext()
A2->B1+C
 >> created A2->B2
 >>> created A2->B2+C
after hasNext()
A2->B2+C
 >> created A2->B3
 >>> created A2->B3+C
after hasNext()
A2->B3+C
 >> created A2->B4
 >> created A4->B0
 >>> created A4->B0+C
after hasNext()
A4->B0+C
 >> created A4->B1
 >>> created A4->B1+C
after hasNext()
A4->B1+C
 >> created A4->B2
 >>> created A4->B2+C
after hasNext()
A4->B2+C
 >> created A4->B3
 >>> created A4->B3+C
after hasNext()
A4->B3+C
 >> created A4->B4

The price to pay is to have A2 and A4 generated up-front, but all the B objects are generated lazily

user_25515800

This is the closest that I can think of which behaves that way you are wanting . I'm putting it here to help with the discussion. In your example, you have two identifiers; for one used when creating A objects, and the other used when creating B objects. With this code, those identifiers are created up-front using the same logic that you had (although I replaced AtomicInteger with IntStream). flatmap is still used, but not at object creation time.

import java.util.Iterator;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class FreeRunner3 {

    public static void main(String[] args) throws InterruptedException {

        Iterator<C> iterator = IntStream.range(1, 5)
            .filter(i -> i % 2 == 0)
            .boxed()
            .flatMap(i -> IntStream.range(1, 5)
                    .filter(j -> j % 2 != 0)
                    .mapToObj(j -> new int[] { i, j }))
            .collect(Collectors.toList())
            .stream()
            .map(id -> new A("A" + id[0]).toB(id[1]))
            .map(B::toC)
            .iterator();

        while (iterator.hasNext()) {
            System.out.println("after hasNext()");
            C next = iterator.next();
            System.out.println(next);
        }

    }

    private static class A {
        private final String name;

        public A(String name) {
            this.name = name;
            System.out.println(" > created " + name);
        }

        public B toB(int i) {
            return new B(this.name + "->B" + i);
        }

        public String toString() {
            return name;
        }
    }

    private static class B {
        private final String name;

        public B(String name) {
            this.name = name;
            System.out.println(" >> created " + name);
        }

        public String toString() {
            return name;
        }

        public C toC() {
            return new C(this.name + "+C");
        }

    }

    private static class C {
        private final String name;

        public C(String name) {
            this.name = name;
            System.out.println(" >>> created " + name);
        }

        public String toString() {
            return name;
        }
    }
}

One difference with this implementation is that an instance of A is created for each instance of B.

 > created A2
 >> created A2->B1
 >>> created A2->B1+C
after hasNext()
A2->B1+C
 > created A2
 >> created A2->B3
 >>> created A2->B3+C
after hasNext()
A2->B3+C
 > created A4
 >> created A4->B1
 >>> created A4->B1+C
after hasNext()
A4->B1+C
 > created A4
 >> created A4->B3
 >>> created A4->B3+C
after hasNext()
A4->B3+C

I thought I would include this as well to help with the discussion (hoping no down votes from anyone). It does the opposite of what you are wanting, eagerly creating all objects up-front before the iterator is called on.

Iterator<C> iterator = IntStream.range(1, 5)
        .filter(i -> i % 2 == 0)
        .mapToObj(i -> new A("A" + i))
        .flatMap(A::getBStream)
        .map(B::toC)
        .collect(Collectors.toList())
        .iterator();

while (iterator.hasNext()) {
    System.out.println("after hasNext()");
    C next = iterator.next();
    System.out.println(next);
}
user_22552170

Instead of creating an iterator and then do something for each element in it, you could simply use a forEach and do it directly in the Stream:

AtomicInteger x = new AtomicInteger();
Stream.generate(() -> null)
        .takeWhile(y -> x.incrementAndGet() < 5)
        .filter(y -> x.get() % 2 == 0)
        .map(n -> new A("A" + x.get()))
        .flatMap(A::getBStream)
        .filter(Objects::nonNull)
        .map(B::toC)
        .forEach(c -> {
            System.out.println("after (no) hasNext()");
            System.out.println(c);
        });

Output:

 > created A2
 >> created A2->B1
 >>> created A2->B1+C
after (no) hasNext()
A2->B1+C
 >> created A2->B3
 >>> created A2->B3+C
after (no) hasNext()
A2->B3+C
 > created A4
 >> created A4->B1
 >>> created A4->B1+C
after (no) hasNext()
A4->B1+C
 >> created A4->B3
 >>> created A4->B3+C
after (no) hasNext()
A4->B3+C

If you need to return the list, instead, you could simply return a Stream:

AtomicInteger x = new AtomicInteger();
Stream<C> stream = Stream.generate(() -> null)
        .takeWhile(y -> x.incrementAndGet() < 5)
        .filter(y -> x.get() % 2 == 0)
        .map(n -> new A("A" + x.get()))
        .flatMap(A::getBStream)
        .filter(Objects::nonNull)
        .map(B::toC);

// return it here

stream.forEach(c -> {
    System.out.println("after (no) hasNext()");
    System.out.println(c);
});

The output would be the same.

user_16542650
Related