suma - ¿Por qué filter() después de flatMap() es "no completamente" vago en las secuencias de Java?

stream java 8 ejemplo

Question

Tengo el siguiente código de muestra:

System.out.println(
       "Result: " +
        Stream.of(1, 2, 3)
                .filter(i -> {
                    System.out.println(i);
                    return true;
                })
                .findFirst()
                .get()
);
System.out.println("-----------");
System.out.println(
       "Result: " +
        Stream.of(1, 2, 3)
                .flatMap(i -> Stream.of(i - 1, i, i + 1))
                .flatMap(i -> Stream.of(i - 1, i, i + 1))
                .filter(i -> {
                    System.out.println(i);
                    return true;
                })
                .findFirst()
                .get()
);

La salida es la siguiente:

1
Result: 1
-----------
-1
0
1
0
1
2
1
2
3
Result: -1

Desde aquí veo que en el primer caso, stream realmente se comporta de manera perezosa: usamos findFirst() por lo que una vez que tenemos el primer elemento, no se invoca nuestro lambda de filtrado. Sin embargo, en el segundo caso que usa flatMap s, vemos que a pesar de que se encuentra el primer elemento que cumple la condición de filtro (es solo cualquier primer elemento, ya que lambda siempre devuelve verdadero), aún se alimentan más contenidos de la secuencia a través de la función de filtrado.

Estoy tratando de entender por qué se comporta así en lugar de darse por vencido después de calcular el primer elemento como en el primer caso. Cualquier información útil sería apreciada.

Answer 1

Con respecto a la rotura con subtransmisiones infinitas, el comportamiento de flatMap se vuelve aún más sorprendente cuando se lanza una operación de cortocircuito intermedia (en lugar de terminal).

Mientras lo siguiente funciona como se esperaba, imprime la secuencia infinita de enteros

Stream.of("x").flatMap(_x -> Stream.iterate(1, i -> i + 1)).forEach(System.out::println);

el siguiente código imprime solo el "1", pero aún no termina:

Stream.of("x").flatMap(_x -> Stream.iterate(1, i -> i + 1)).limit(1).forEach(System.out::println);

No puedo imaginar una lectura de la especificación en la que eso no era un error.

Answer 2

Desafortunadamente .flatMap() no es vago. Sin embargo, una solución personalizada de flatMap está disponible aquí: por qué .flatMap () es tan ineficiente (no vago) en java 8 y java 9

Answer 3

En mi biblioteca gratuita de StreamEx introduje los colectores de cortocircuito. Al recopilar secuencia secuencial con un colector de cortocircuito (como MoreCollectors.first() ) exactamente se consume un elemento de la fuente. Internamente se implementa de una manera bastante sucia: usando una excepción personalizada para romper el flujo de control. Usando mi biblioteca, su muestra podría reescribirse de esta manera:

System.out.println(
        "Result: " +
                StreamEx.of(1, 2, 3)
                .flatMap(i -> Stream.of(i - 1, i, i + 1))
                .flatMap(i -> Stream.of(i - 1, i, i + 1))
                .filter(i -> {
                    System.out.println(i);
                    return true;
                })
                .collect(MoreCollectors.first())
                .get()
        );

El resultado es el siguiente:

-1
Result: -1

Answer 4

Estoy de acuerdo con otras personas, este es un error abierto en bugs.openjdk.java.net/browse/JDK-8075939 . Y dado que aún no se ha solucionado más de un año después. Me gustaría recomendarte: AbacusUtil

N.println("Result: " + Stream.of(1, 2, 3).peek(N::println).first().get());

N.println("-----------");

N.println("Result: " + Stream.of(1, 2, 3)
                        .flatMap(i -> Stream.of(i - 1, i, i + 1))
                        .flatMap(i -> Stream.of(i - 1, i, i + 1))
                        .peek(N::println).first().get());

// output:
// 1
// Result: 1
// -----------
// -1
// Result: -1

Divulgación: Soy el desarrollador de AbacusUtil.

Answer 5

Los elementos de la secuencia de entrada se consumen perezosamente uno por uno. El primer elemento, 1 , es transformado por los dos flatMap s en la secuencia -1, 0, 1, 0, 1, 2, 1, 2, 3 , de modo que toda la secuencia corresponde solo al primer elemento de entrada. Las corrientes anidadas se materializan con entusiasmo por la tubería, luego se aplanan y luego se alimentan a la etapa de filter . Esto explica su salida.

Lo anterior no se deriva de una limitación fundamental, pero probablemente haría las cosas mucho más complicadas para conseguir la holgazanería completa de las secuencias anidadas. Sospecho que sería un desafío aún mayor hacer que funcione. A modo de comparación, las secciones perezosas de Clojure obtienen otra capa de envoltura para cada nivel de anidamiento. Debido a este diseño, las operaciones pueden incluso fallar con Error cuando la anidación se ejerce al extremo.

Answer 6

TL; DR, esto se ha solucionado en bugs.openjdk.java.net/browse/JDK-8075939 y se ha corregido en Java 10 (y se ha trasportado a Java 8 en JDK-8225328 ).

Al analizar la implementación ( ReferencePipeline.java ) vemos el método [ link ]

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

que se invocará para la operación findFirst . Lo especial a tener en cuenta es el sink.cancellationRequested() que permite finalizar el ciclo en la primera coincidencia. Comparar con [ 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);
                    }
                }
            };
        }
    };
}

El método para avanzar un elemento termina llamando a forEach en la forEach sin ninguna posibilidad de terminación anticipada y el comentario al comienzo del método flatMap incluso habla sobre esta característica ausente.

Dado que esto es más que una cuestión de optimización, ya que implica que el código simplemente se rompe cuando la transmisión secundaria es infinita, espero que los desarrolladores pronto demuestren que "pueden hacerlo mejor que esto" ...

Para ilustrar las implicaciones, mientras Stream.iterate(0, i->i+1).findFirst() funciona como se esperaba, Stream.of("").flatMap(x->Stream.iterate(0, i->i+1)).findFirst() terminará en un bucle infinito.

En cuanto a la especificación, la mayor parte se puede encontrar en el

capítulo "Operaciones de transmisión y canalizaciones" de la especificación del paquete :

...

Las operaciones intermedias devuelven una nueva secuencia. Siempre son vagos ;

...

… La pereza también permite evitar examinar todos los datos cuando no es necesario; para operaciones como "buscar la primera cadena de más de 1000 caracteres", solo es necesario examinar las cadenas suficientes para encontrar una que tenga las características deseadas sin examinar todas las cadenas disponibles en la fuente. (Este comportamiento se vuelve aún más importante cuando el flujo de entrada es infinito y no simplemente grande).

...

Además, algunas operaciones se consideran operaciones de cortocircuito . Una operación intermedia está en cortocircuito si, cuando se presenta con entrada infinita, puede producir una corriente finita como resultado. Una operación de terminal está en cortocircuito si, cuando se presenta con una entrada infinita, puede terminar en un tiempo finito. Tener una operación de cortocircuito en la tubería es una condición necesaria, pero no suficiente, para que el procesamiento de una corriente infinita termine normalmente en un tiempo finito.

Está claro que una operación de cortocircuito no garantiza una terminación de tiempo finito, por ejemplo, cuando un filtro no coincide con ningún elemento que el procesamiento no puede completar, pero una implementación que no admite ninguna terminación en tiempo finito simplemente ignorando La naturaleza de cortocircuito de una operación está muy lejos de la especificación.