Why filter() after flatMap() is “not completely” lazy in Java streams?

by

TL;DR, this has been addressed in JDK-8075939 and fixed in Java 10 (and backported to Java 8 in JDK-8225328).

When looking into the implementation (ReferencePipeline.java) we see the method [link]

@Override
final void forEachWithCancel(Spliterator<P_OUT> spliterator, Sink<P_OUT> sink) {
    do { } while (!sink.cancellationRequested() && spliterator.tryAdvance(sink));
}

which will be invoke for findFirst operation. The special thing to take care about is the sink.cancellationRequested() which allows to end the loop on the first match. Compare to [link]

@Override
public final <R> Stream<R> flatMap(Function<? super P_OUT, ? extends Stream<? extends R>> mapper) {
    Objects.requireNonNull(mapper);
    // We can do better than this, by polling cancellationRequested when stream is infinite
    return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
                                 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
            return new Sink.ChainedReference<P_OUT, R>(sink) {
                @Override
                public void begin(long size) {
                    downstream.begin(-1);
                }

                @Override
                public void accept(P_OUT u) {
                    try (Stream<? extends R> result = mapper.apply(u)) {
                        // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
                        if (result != null)
                            result.sequential().forEach(downstream);
                    }
                }
            };
        }
    };
}

The method for advancing one item ends up calling forEach on the sub-stream without any possibility for earlier termination and the comment at the beginning of the flatMap method even tells about this absent feature.

Since this is more than just an optimization thing as it implies that the code simply breaks when the sub-stream is infinite, I hope that the developers soon prove that they “can do better than this”…

To illustrate the implications, while Stream.iterate(0, i->i+1).findFirst() works as expected, Stream.of("").flatMap(x->Stream.iterate(0, i->i+1)).findFirst() will end up in an infinite loop.

Regarding the specification, most of it can be found in the

chapter “Stream operations and pipelines” of the package specification:

…

Intermediate operations return a new stream. They are always lazy;

…

… Laziness also allows avoiding examining all the data when it is not necessary; for operations such as “find the first string longer than 1000 characters”, it is only necessary to examine just enough strings to find one that has the desired characteristics without examining all of the strings available from the source. (This behavior becomes even more important when the input stream is infinite and not merely large.)

…

Further, some operations are deemed short-circuiting operations. An intermediate operation is short-circuiting if, when presented with infinite input, it may produce a finite stream as a result. A terminal operation is short-circuiting if, when presented with infinite input, it may terminate in finite time. Having a short-circuiting operation in the pipeline is a necessary, but not sufficient, condition for the processing of an infinite stream to terminate normally in finite time.

It’s clear that a short-circuiting operation doesn’t guaranty a finite time termination, e.g. when a filter doesn’t match any item the processing can’t complete, but an implementation which doesn’t support any termination in finite time by simply ignoring the short-circuiting nature of an operation is far off the specification.

More Related Contents:

  1. Java 8 List into Map
  2. How can I throw CHECKED exceptions from inside Java 8 streams?
  3. How to force max to return ALL maximum values in a Java Stream?
  4. Java 8 lambdas, Function.identity() or t->t
  5. Copy a stream to avoid “stream has already been operated upon or closed”
  6. Java 8 Nested (Multi level) group by
  7. Using Java 8’s Optional with Stream::flatMap
  8. Java 8 Streams: multiple filters vs. complex condition
  9. Group by and sum objects like in SQL with Java lambdas?
  10. Retrieving a List from a java.util.stream.Stream in Java 8
  11. Filter Java Stream to 1 and only 1 element
  12. Java 8, Streams to find the duplicate elements
  13. Why doesn’t java.util.Collection implement the new Stream interface?
  14. Find maximum, minimum, sum and average of a list in Java 8
  15. Does Java SE 8 have Pairs or Tuples?
  16. How to use a Java8 lambda to sort a stream in reverse order?
  17. Java 8 Collectors.toMap SortedMap
  18. Why does Stream.allMatch() return true for an empty stream?
  19. Java 8 grouping using custom collector?
  20. How can I throw CHECKED exceptions from inside Java 8 lambdas/streams?
  21. How to apply multiple predicates to a java.util.Stream?
  22. Parallel stream from a HashSet doesn’t run in parallel
  23. Picking elements of a list until condition is met with Java 8 Lambdas
  24. Non-interference exact meaning in Java 8 streams
  25. In Java 8 how do I transform a Map to another Map using a lambda?
  26. Convert String array to Map using Java 8 Lambda expressions
  27. Group sequences of values
  28. Java 8 stream map to list of keys sorted by values
  29. Nested Java 8 parallel forEach loop perform poor. Is this behavior expected?
  30. Parallel flatMap always sequential

Leave a Comment