Segmented repartition

[mpilquist]

The PR builds off of #950 and implements repartition such that each segment in the underlying stream is scanned instead of unconsing single elements.

Note I had to change the repartition function to return a Chunk instead of a Segment but this is okay -- repartitioning a single element should not require an expansion in to an infinite segment.

@soumyarupsarkar -- please take a look. This uses the double layer scanning algorithm I attempted to describe in one of the reviews of #950.

[soumyarupsarkar]

I see, this makes sense. Thanks!

[soumyarupsarkar]

You may want to add this as a test case with an infinite segment in the stream: Stream.segment(Segment.from(0).map(_.toInt)).repartition(x => Chunk(x, x)).take(4).toVector shouldBe Vector(0, 1, 3, 6)

[mpilquist]

@soumyarupsarkar Thanks for spotting that. It turns out that this is a limitation when working with infinite streams and pipelines that cause less than 1024 elements to be emitted. For example, here's a program that also hangs:

Stream.segment(Segment.from(0L)).filter(_ < 2).take(2).toVector

Conceptually, the take(2) is satisfied by the first two elements from the infinite segment, and so we should be able to terminate. However, the stream interpreter will wait for 1024 elements to be collected as a result of the filter before moving on. To fix, I think we'd need to change Segment#splitAt to take a maxSteps argument or something, which tracks the amount of work the segment is doing in creating the 1024 elements.

[mpilquist]

@soumyarupsarkar On second thought, the following program will never emit any elements:

Stream.segment(Segment.from(0).map(_.toInt)).repartition(x => Chunk(x + 2)).take(1).toVector

Since each input element yields a single chunk, the carry value ends up representing a fold over all input elements and we never output anything. So while the example I gave earlier is in fact an issue with infinite streams, the repartition hang isn't an example of that issue.