Stream Pipelines, Laziness and Parallelism

Laziness in depth

Laziness means intermediate operations merely build a plan — no data flows until a terminal operation triggers execution. This is not just a performance detail; it fundamentally changes what is possible.

Benefit 1: Short-circuiting

Operations like limit(n) or findFirst() can stop the pipeline early, avoiding unnecessary work:

List<String> names = List.of("Alice", "Bob", "Charlie", /* ... million more ... */);

Optional<String> firstLongName = names.stream()
    .filter(n -> n.length() > 10)
    .findFirst();

With lazy execution, the pipeline checks each element: “Alice” (no), “Bob” (no), and so on until the first match is found. It does not filter all million names. With eager (external) iteration, you would need to manually break out of the loop.

// Infinite stream — only possible because of laziness
Stream.iterate(0, x -> x + 1)       // 0, 1, 2, 3, ... infinite
    .filter(x -> x % 2 == 0)        // keep evens
    .map(x -> x * x)                // square them
    .limit(10)                      // take first 10 results
    .forEach(System.out::println);  // print: 0, 4, 16, 36, 64, 100, 144, 196, 256, 324

Without laziness, Stream.iterate(0, x -> x + 1) would try to generate all integers — impossible. Laziness means elements are generated on demand, and limit(10) stops the pipeline after 10 results.

Benefit 2: Operation fusion

The stream library can combine multiple operations into a single pass over the data:

names.stream()
    .filter(n -> n.startsWith("A"))
    .map(String::toUpperCase)
    .collect(Collectors.toList());

In external iteration, you would filter in one loop, collect, then map in another loop — two passes. The stream library can fuse filter and map into a single traversal: for each element, check the predicate; if it passes, apply the mapping function; then pass the result downstream.

Benefit 3: No intermediate storage

Stream operations create intermediate results lazily and pass them downstream. There is no list of filtered results, then a list of mapped results — each element flows through the entire pipeline before the next element is processed (for sequential streams). This avoids unnecessary memory allocation.

How laziness affects execution order

For sequential streams, elements are processed one at a time through the entire pipeline:

List.of("Alice", "Bob", "Charlie").stream()
    .peek(n -> System.out.println("Filtering: " + n))
    .filter(n -> n.length() > 3)
    .peek(n -> System.out.println("  Mapping: " + n))
    .map(String::toUpperCase)
    .forEach(n -> System.out.println("    Result: " + n));

Output (element-by-element, not phase-by-phase):

Filtering: Alice
  Mapping: Alice
    Result: ALICE
Filtering: Bob
Filtering: Charlie
  Mapping: Charlie
    Result: CHARLIE

Note: “Bob” is filtered out and never mapped. Elements do NOT all go through filter, then all go through map.

Exception: Stateful operations like sorted() force all upstream elements to be buffered before any element passes downstream. sorted() is a barrier in the pipeline.

Parallel streams

Parallel streams split the source into substreams, process them concurrently in the common fork-join pool, and merge results:

// Sequential
long count = words.stream()
    .filter(w -> w.length() > 10)
    .count();

// Parallel — same pipeline
long count = words.parallelStream()
    .filter(w -> w.length() > 10)
    .count();

Requirements for correct parallelisation

For a parallel stream to be correct, the operations must be:

1. Stateless — Each element’s processing does not depend on other elements or on mutable state outside the lambda.

2. Non-interfering — The operations do not modify the source collection during processing.

3. Associative (for reductions) — For reduce, the combining operation must be associative, because the order of combining substream results is not guaranteed.

// UNSAFE: mutable state in lambda
List<String> results = new ArrayList<>();
names.parallelStream()
    .filter(n -> n.length() > 3)
    .forEach(n -> results.add(n));  // concurrent modification of ArrayList!

// SAFE: use collect
List<String> results = names.parallelStream()
    .filter(n -> n.length() > 3)
    .collect(Collectors.toList());  // thread-safe collector

When to use parallel streams

• The pipeline is CPU-intensive (not I/O-bound).
• The source is large (thousands of elements — parallelism overhead is significant for small collections).
• Operations are stateless and non-interfering.
• You are using a thread-safe collector (like Collectors.toList()).

When NOT to use parallel streams:

• Small collections (overhead dominates).
• I/O-bound operations (parallelism on CPU cores does not help).
• Operations with side effects.
• Operations that rely on encounter order (e.g., findFirst() on a parallel stream still respects encounter order, but the cost may outweigh the benefit).

stream() vs parallelStream()

Prefer stream() by default. Only use parallelStream() when you can measure a benefit. The Tripos expects you to understand when parallelism is appropriate, not to default to it.

Streams and immutability

Stream operations never mutate the source collection. Each step produces a conceptual new stream or result. This dovetails with functional programming principles:

List<String> original = new ArrayList<>(List.of("a", "b", "c"));
List<String> result = original.stream()
    .map(String::toUpperCase)
    .collect(Collectors.toList());

// original is still ["a", "b", "c"]
// result is ["A", "B", "C"]

This immutability makes reasoning about code easier (no surprises from side effects) and enables safe parallelisation (no data races on the source).

Common Tripos comparison

A typical Tripos question asks you to show a sequential loop version, then refactor to a stream pipeline, and then discuss whether parallelisation is appropriate.

External iteration version:

int count = 0;
for (String s : words) {
    if (s.length() > 5 && s.startsWith("A")) {
        count++;
    }
}

Stream pipeline refactoring:

long count = words.stream()
    .filter(s -> s.length() > 5)
    .filter(s -> s.startsWith("A"))
    .count();

Parallelisation discussion:

• This pipeline is stateless and non-interfering — it is safe to parallelise.
• But count() is cheap and the source may be small — the parallelism overhead may outweigh the benefit.
• parallelStream() would be justified if words had millions of elements and the predicates were expensive.

Tripos traps

1. Assuming streams are always faster — Streams have overhead. For small collections, a simple for-loop is often faster. Streams are about readability and composability, not raw performance.
2. Stateful lambdas in parallel streams — A peek that increments a counter, a filter that reads a shared variable. These are bugs waiting to happen in parallel pipelines.
3. Ordering assumptions — forEach does not guarantee order for parallel streams. Use forEachOrdered if encounter order matters.
4. Infinite stream without short-circuiting — Stream.iterate(0, x -> x + 1).collect(Collectors.toList()) runs forever. Always ensure an infinite stream has a limit, findFirst, or other short-circuiting operation.