 
   defp get_first_row(stream, options) do
     row = stream
-            |> LineAggregator.aggregate(options |> Enum.into([]))
+            |> LineAggregator.aggregate(options)
             |> Enum.take(1)
             |> List.first
+
     case decode_row({ row, 0 }, options) do
       { :ok, data } -> data
       _ -> []


  defp options_with_defaults(options) do
    num_pipes = options |> Keyword.get(:num_pipes, Defaults.num_workers)
    options
    |> Keyword.merge(num_pipes: num_pipes,
                     num_workers: options |> Keyword.get(:num_workers, num_pipes),
                     multiline_escape: options |> Keyword.get(:multiline_escape, true),
                     headers: options |> Keyword.get(:headers, false),
                     row_length: false)
    |> Enum.into(%{})
  end

  defp prepare_headers({ stream, %{ headers: true } = options }) do
    { stream |> Stream.drop(1),
      %{ options | :headers => get_first_row(stream, options) } }
  end
  defp prepare_headers(job), do: job

  defp prepare_row_length({ stream, %{ headers: headers } = options}) do
    first_row = if headers, do: headers, else: stream |> get_first_row(options)
    { stream, %{ options | :row_length => Enum.count(first_row) } }
  end

  defp validate_row_length({ data, _}, false), do: { :ok, data }

    |> decode_stream
  end

  defp decode_stream({ stream, options }) do
    multiline_escape = options |> Keyword.get(:multiline_escape)

    stream
    |> aggregate(multiline_escape)
    |> Stream.with_index
    |> ParallelStream.map(&(decode_row(&1, options)), options)
  end

  defp decode_row({ nil, 0 }, options) do
    { :ok, [] }
  end
  defp decode_row({ line, index }, options) do
    headers = options |> Keyword.get(:headers)
    row_length = options |> Keyword.get(:row_length)

    with { :ok, parsed, _ } <- parse_row({ line, index }, options),
    { :ok, _ } <- validate_row_length({ parsed, index }, row_length),
    do: build_row(parsed, headers)
  end

  #defp decode_row({ nil, 0 }, %{ row_length: false }), do: { :ok, [] }

  defp parse_row({ line, index}, options) do
    with { :ok, lex, _ } <- Lexer.lex({ line, index }, options),
