Reduce Stream to Contain Last Element Only

[Ron McLeod]

I want to be able to transform a stream with zero or more elements into a stream with zero or one elements, where the element is the last element from the original stream.

The code below does work, but I wondering if there might be a better way where I don't need to process each element one-by-one.

public class LastElementTest {
    @Test
    void remove_all_but_last_element() {
        var stream = Stream.of("first", "middle", "last");
        var lastStream = streamWithLastElement(stream);
        lastStream.forEach(System.out::println);
    }

    <T> Stream<T> streamWithLastElement(Stream<T> in) {
        return Stream.ofNullable(in.reduce((acc, ele) -> ele).orElse(null));
    }
}

[Piet Souris]

First thought: intstream iterate from size - 1 to 0, map to the list and limit 1

[Ron McLeod]

Piet Souris wrote:First thought: intstream iterate from size ...

With my case, I don't know the size in-advance, and to discover it, I would need to fully consume the stream.

[Piet Souris]

Can't think of anything useful, so your reduce looks fine, but beware of infinite streams

[Mike Simmons]

One minor tweak is, there's a shorter way to express the same thing:

  <T> Stream<T> streamWithLastElement2(Stream<T> in) {
    return in.reduce((acc, ele) -> ele).stream();
  }

While it's a little more concise, it does have the same issue that the original did, in the sense that it looks at the entire stream.  There's not a good built-in solution that I found.  But, I did manage to come up with a solution that seems to work:

  public static <T> Stream<T> streamWithLastElement(Stream<T> in) {
    return lastElementFrom(in.parallel().spliterator()).stream();
  }

  private static <T> Optional<T> lastElementFrom(Spliterator<T> split) {
    var subsplit = split.trySplit();
    if (subsplit != null) {
      return lastElementFrom(split).or(() -> lastElementFrom(subsplit));
    } else {
      var last = new AtomicReference<T>();
      split.forEachRemaining(t -> last.set(t));
      return Optional.ofNullable((T) last.get());
    }
  }

The parallel() seems to be necessary, at least in my tests, to get the stream to return a Spliterator that will actually do a split.  Otherwise I kept getting the split() returning null, which means there are no further subdivisions possible.  Interestingly, we're not actually using multiple threads to evaluate the spliterator.  We're just telling the stream to act as if we could use parallel streams, thereby encouraging the split() to allow us to split off some of elements, even if it thinks it could do this faster in a single thread.

To better understand how it works and to experiment with it (and verify that it really can be much faster than reading everything):

  public static void main(String[] args) {
    // Deliberately setting up a very big, very slow stream...
    var stream = IntStream.range(1, 1_000_000_000).mapToObj(n -> {
      System.out.println("Evaluating element " + n + " in thread " + Thread.currentThread().getName());
      try {
        Thread.sleep(1000);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
      return n;
    });
    var lastStream = streamWithLastElement(stream);
    System.out.println("Result: " + lastStream.toList());
  }

  public static <T> Stream<T> streamWithLastElement(Stream<T> in) {
    return lastElementFrom(in.parallel().spliterator()).stream();
  }

  private static <T> Optional<T> lastElementFrom(Spliterator<T> split) {
    var subsplit = split.trySplit();
    if (subsplit != null) {
      System.out.println("split: " + split.estimateSize() + ", subsplit: " + subsplit.estimateSize());
      return lastElementFrom(split).or(() -> lastElementFrom(subsplit));
    } else {
      System.out.println("No further split, size=" + split.estimateSize());
      var last = new AtomicReference<T>();
      split.forEachRemaining(t -> last.set(t));
      return Optional.ofNullable((T) last.get());
    }
  }

Output:

split: 875000000, subsplit: 124999999
split: 765625000, subsplit: 109375000
split: 669921875, subsplit: 95703125
split: 586181641, subsplit: 83740234
split: 512908936, subsplit: 73272705
split: 448795319, subsplit: 64113617
split: 392695905, subsplit: 56099414
split: 343608917, subsplit: 49086988
split: 300657803, subsplit: 42951114
split: 263075578, subsplit: 37582225
split: 230191131, subsplit: 32884447
split: 201417240, subsplit: 28773891
split: 176240085, subsplit: 25177155
split: 154210075, subsplit: 22030010
split: 134933816, subsplit: 19276259
split: 118067089, subsplit: 16866727
split: 103308703, subsplit: 14758386
split: 90395116, subsplit: 12913587
split: 79095727, subsplit: 11299389
split: 69208762, subsplit: 9886965
split: 60557667, subsplit: 8651095
split: 52987959, subsplit: 7569708
split: 46364465, subsplit: 6623494
split: 40568907, subsplit: 5795558
split: 35497794, subsplit: 5071113
split: 31060570, subsplit: 4437224
split: 27177999, subsplit: 3882571
split: 23780750, subsplit: 3397249
split: 20808157, subsplit: 2972593
split: 18207138, subsplit: 2601019
split: 15931246, subsplit: 2275892
split: 7965623, subsplit: 7965623
split: 3982812, subsplit: 3982811
split: 1991406, subsplit: 1991406
split: 995703, subsplit: 995703
split: 497852, subsplit: 497851
split: 248926, subsplit: 248926
split: 124463, subsplit: 124463
split: 62232, subsplit: 62231
split: 31116, subsplit: 31116
split: 15558, subsplit: 15558
split: 7779, subsplit: 7779
split: 3890, subsplit: 3889
split: 1945, subsplit: 1945
split: 973, subsplit: 972
split: 487, subsplit: 486
split: 244, subsplit: 243
split: 122, subsplit: 122
split: 61, subsplit: 61
split: 31, subsplit: 30
split: 16, subsplit: 15
split: 8, subsplit: 8
split: 4, subsplit: 4
split: 2, subsplit: 2
split: 1, subsplit: 1
No further split, size=1
Evaluating element 999999999 in thread main
Result: [999999999]

Note that it only evaluates the last element.  For some streams, it might end up having to evaluate more, in cases where it's hard for the Spliterator to estimate size correctly and some splits might not contain any actual elements.

[Carey Brown]

This has been enlightening. Here's both together with some timing.

import java.util.Optional;
import java.util.Spliterator;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.IntStream;
import java.util.stream.Stream;

public class StreamLastElement
{

	public static void main( String[] args )
	{
		long		start, stop;
		
		start = System.currentTimeMillis();
		usingSpliterator();
		stop = System.currentTimeMillis();
		System.out.printf( "%-12s %s%n", "Spliterator", duration(start,stop) );
		
		start = System.currentTimeMillis();
		sequential();
		stop = System.currentTimeMillis();
		System.out.printf( "%-12s %s%n", "Sequential", duration(start,stop) );
	}

	public static void usingSpliterator()
	{
		// Deliberately setting up a very big, very slow stream...
		var stream = IntStream.range( 1, 1_000_000_000 ).mapToObj( n -> {
//			System.out.println( "Evaluating element " + n + " in thread "
//					+ Thread.currentThread().getName() );
//			try
//			{
//				Thread.sleep( 1000 );
//			}
//			catch( InterruptedException e )
//			{
//				throw new RuntimeException( e );
//			}
			return n;
		} );
		var lastStream = streamWithLastElement( stream );
		System.out.println( "Result: " + lastStream.toList() );
	}

	public static <T> Stream<T> streamWithLastElement( Stream<T> in )
	{
		return lastElementFrom( in.parallel().spliterator() ).stream();
	}

	private static <T> Optional<T> lastElementFrom( Spliterator<T> split )
	{
		var subsplit = split.trySplit();
		if( subsplit != null )
		{
//			System.out.println( "split: " + split.estimateSize()
//					+ ", subsplit: " + subsplit.estimateSize() );
			return lastElementFrom( split )
					.or( () -> lastElementFrom( subsplit ) );
		}
		else
		{
//			System.out.println(
//					"No further split, size=" + split.estimateSize() );
			var last = new AtomicReference<T>();
			split.forEachRemaining( t -> last.set( t ) );
			return Optional.ofNullable( (T)last.get() );
		}
	}

	public static void sequential()
	{
		var stream = IntStream.range( 1, 1_000_000_000 ).mapToObj( n -> n );
//		var stream = Stream.of( "first", "middle", "last" );
		var lastStream = streamWithLastElement2( stream );
		System.out.println( "Result: " + lastStream.toList() );
	}

	public static <T> Stream<T> streamWithLastElement2( Stream<T> in )
	{
		return Stream
				.ofNullable( in.reduce( ( acc, ele ) -> ele ).orElse( null ) );
	}
	
	private static String duration( long start, long stop )
	{
		long	millis = stop - start;
		long	hours = millis / (60L * 60L * 1_000L);
		millis %= (60L * 60L * 1_000L);
		
		long	minutes = millis / (60L * 1_000L);
		millis %= (60L * 1_000L);
		
		long	seconds = millis / (1_000L);
		millis %= 1_000L;
		
		return String.format( "%02d:%02d:%02d.%03d", hours, minutes, seconds, millis );
	}

}

Output:

Result: [999999999]
Spliterator  00:00:00.007
Result: [999999999]
Sequential   00:00:02.547

[Piet Souris]

Brilliant, Mike, have a cow! Never worked with spliterators, but this is a good starting place.

[Rob Spoor]

Be warned that this trySplit is only guaranteed to work if the spliterator is ordered:

https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/Spliterator.html#trySplit() wrote:
If this spliterator can be partitioned, returns a Spliterator covering elements, that will, upon return from this method, not be covered by this Spliterator.

If this Spliterator is ORDERED, the returned Spliterator must cover a strict prefix of the elements.

(On the other hand, if the spliterator isn't ordered, then the stream probably wasn't either, and there's no real definition of "last").

It also won't work on infinite streams, but then again, no solution would work on infinite streams as these don't have a last element.

Unless this Spliterator covers an infinite number of elements, repeated calls to trySplit() must eventually return null.

[Mike Simmons]

Yeah, I was thinking about those two issues as well.  I don't know anything to be done about an infinite stream - that's a common problem though.  People creating infinite streams need to get used to ctrl-C. . To be fair, I quite like using infinite streams as building blocks - an infinite stream of primes, for example... with the understanding that you need to make sure there's a way to terminate eventually, when you call a terminal operation.

As for ordering though, we can check that, and throw an informative error:

  public static <T> Stream<T> streamWithLastElement(Stream<T> in) {
    var split = in.parallel().spliterator();
    if (!split.hasCharacteristics(Spliterator.ORDERED)) {
      throw new IllegalArgumentException("Can't get last element of unordered stream");
    }
    return lastElementFrom(split).stream();
  }

Unless there's ever a chance that we might want to process a stream not actually marked as ordered, but which nonetheless behaves in an ordered fashion.  Don't know if that's actually possible, perhaps not.  But if someone wanted to try it anyway, they could at least log a warning, rather than throw an exception.