Calling `decode` with `num_pipes: 1` multiple times on the same stream yields different orderings each time

[adamesque]

I'm noticing some weird stuff happening with the decoder when I call it multiple times in an IEx session.

Given a file my_file.csv with ~200 lines, doing

File.stream!("my_file.csv") |> CSV.Decoder.decode(num_pipes: 1) |> Enum.take(2)

yields the first 2 lines of the file the first time I run it, but subsequent invocations yield seemingly random / out-of-order lines.

I also get strange behavior when working with a simple stream (although I'm not able to reproduce the ordering issue):

iex(1)> stream = ~w(1,2,3,4 5,6,7,8 9,10,11,12 13,14,15,16) |> Stream.map(&(&1))
#Stream<[enum: ["1,2,3,4", "5,6,7,8"],
 funs: [#Function<45.113986093/1 in Stream.map/2>]]>
iex(2)> stream |> Enum.take(2)
["1,2,3,4", "5,6,7,8"]
iex(3)> stream |> Enum.take(2)
["1,2,3,4", "5,6,7,8"]
iex(4)> CSV.Decoder.decode(stream, num_pipes: 1) |> Enum.take(2)
#Function<25.113986093/2 in Stream.resource/3>
iex(5)> CSV.Decoder.decode(stream, num_pipes: 1) |> Enum.take(2)
[["1", "2", "3", "4"], ["5", "6", "7", "8"]]
iex(6)> CSV.Decoder.decode(stream, num_pipes: 1) |> Enum.take(2)
[]
iex(7)> CSV.Decoder.decode(stream, num_pipes: 1) |> Enum.take(2)
[["1", "2", "3", "4"], ["5", "6", "7", "8"]]
iex(8)> CSV.Decoder.decode(stream, num_pipes: 1) |> Enum.take(2)
[]

[beatrichartz]

This has to do with the partial take that is applied - there is some state left over in the processes when the stream is not entirely drained.

Thanks for raising, I will provide a fix for this bug tomorrow.

[beatrichartz]

Ok, this should be fixed right now - could you try the version in master with your own file and confirm?

btw a better way to write this in this case would be:

stream |> Stream.take(2) |> CSV.Decoder.decode(num_pipes: 1) |> Enum.into([])

This still only takes a certain amount from your stream whilst not unnecessarily invoking parsing of rows.

[beatrichartz]

Feel free to reopen if the bug persists in master.