[C++] Add Byte Range to CSV Reader ReadOptions

[asfimport]

Sometimes it's desirable to just read a portion of a CSV. The best way to do that is to pass in a list of byte ranges to CSV read options that specify where in the CSV you want to read. These byte ranges don't necessarily have to be aligned on line break boundaries, the CSV reader should just read until the end of the line, and skip anything before the first line break in a byte range.  

Based on discussion, the scope is going to be reduced here. The first implementation will support a single byte range that is already assumed to be aligned on byte boundaries.

Will not handle quotes/returns and other edge cases.

Reporter: Ziheng Wang / @marsupialtail

PRs and other links:

• GitHub Pull Request #13830
• GitHub Pull Request #14078

_{Note: This issue was originally created as ARROW-17313. Please see the migration documentation for further details.}

[asfimport]

Weston Pace / @westonpace:
For additional motivation, this overlaps with the way Substrait expresses partitioning information. Substrait allows any file type to include "start byte" and "length" to slice the file. For file types like parquet & IPC this would involve grabbing all row groups whose first byte falls in that range (even though this may mean reading beyond the end of the specified range). The advantage is that there is then a uniform API for partitioning files across formats.

Another advantage here is that this would allow us to potentially parallelize chunking at a minor cost of overreading a bit for each block. This overreading could be avoided if we knew we were going to read multiple blocks. For example, if we know we want to read blocks 20-30 then we issue reads for blocks 20-31. As soon as any two consecutive blocks are loaded we can start parsing the lower block of the pair.

So the algorithm for each block boils down to:

Although...now that I type this up...I remember a potential flaw in this logic. Finding the "first line delimiter" in a block can be an impossible problem if newlines are allowed inside of delimiters. Though maybe we don't need to support that case, I don't recall.

CC @pitrou for additional thoughts

[asfimport]

Ziheng Wang / @marsupialtail:
My proposal is that we will allow additional fields in ScanOptions that specifies the byte ranges to read for each fragment. 

Those byte ranges will be updated when you are calling OpenReaderAsync to be aligned to line breaks, potentially in another async function that samples the file around the byte range boundaries and figure out where the line breaks are. 

Then these aligned byte ranges will be used to create a MaskedRandomAccessFile object, which is a new object that resembles the RandomAcessFile object with all of its interface, except it will skip bytes that it's not supposed to read using seek. This will not read those skip-bytes on either disk or network based files. 

We pass this MaskedRandomAccessFile object to make a BufferedInputStream and a StreamingReader without any further change in code. The CSV StreamingReader has no idea that it is only reading partial chunks in the underlying file.

The alternative to deal with this in the CSV StreamingReaderImpl. However this is very complicated as it can only access a BufferedInputStream which is not seekable. Adding seek functionality to InputStream probably doesn't make sense when the underlying InputStream is not a file. 

[asfimport]

Yibo Cai / @cyb70289:
I have same concern as Weston. CSV parsing is stateful. AFAIK, figuring out the line break has to be done in sequential, if we support "quote" or "escape" or custimized delimiter, etc.
Some examples:

• The sample block starts inside a "quoted" field
• The first char of a block is "\n" but the last char of previous block is an "escape"
• Sample at middle of "\r\n" may also be confusing

[asfimport]

Ziheng Wang / @marsupialtail:
I think if you support things like this, then it's impossible to parallel read a CSV file.

[asfimport]

Antoine Pitrou / @pitrou:
It's not too late to change the Substrait spec, is it?

Or we can raise NotImplemented if the offset is ever-non-zero.

[asfimport]

Antoine Pitrou / @pitrou:
That said, it can also be implemented in a "dumb" way: call RandomAccessFile::GetStream to get a subset of the input file, and let the CSV reader try to make sense of it.

[asfimport]

Ziheng Wang / @marsupialtail:
@pitrou can you elaborate a bit on your way?

[asfimport]

Antoine Pitrou / @pitrou:
There's not much to elaborate. RandomAccessFile::GetStream gives you a stream over a subrange of a file. See API docs:
https://arrow.apache.org/docs/cpp/api/io.html#_CPPv4N5arrow2io16RandomAccessFile9GetStreamENSt10shared_ptrI16RandomAccessFileEE7int64_t7int64_t

[asfimport]

Ziheng Wang / @marsupialtail:
Ah I meant what we should do about the linbreaks and quotes etc.

[asfimport]

Antoine Pitrou / @pitrou:
Nothing. The Substrait producer should produce valid ranges. There is no reason to accept random invalid ranges.

[asfimport]

Weston Pace / @westonpace:

It's not too late to change the Substrait spec, is it?
Or we can raise NotImplemented if the offset is ever-non-zero.

Raising "not implemented" in this case is fine I'm sure. If it can't be done then it can't be done. Perhaps we can avoid most of these cases by reading a little bit (e.g. 32 bytes) before the beginning of the block as well.

The sample block starts inside a "quoted" field

I think this is only a problem if we allow newlines in values. We should reject a partial read if newlines_in_values is false.

The first char of a block is "\n" but the last char of previous block is an "escape"

Reading a bit early would help here as long as it isn't a really long chain of escapes which should be rare and detectable (we could error in this case).

Sample at middle of "\r\n" may also be confusing

Reading a bit early would help here too.

[asfimport]

Weston Pace / @westonpace:

We should reject a partial read if newlines_in_values is false.

Oops, I mean, if newlines_in_values is true :)

[asfimport]

Ziheng Wang / @marsupialtail:
Also this will not support compressed formats, at least at the start

[asfimport]

Ziheng Wang / @marsupialtail:
Ideally we update the Dataset Scanner to be able to take in different byte ranges for different fragments. Or is this not required?

A complication would be that fragments right now don't seem to have some sort of "ID", so it might be hard for a user to specify which fragments should read which byte ranges. The way to do this would be to let the user pass in a dict in the ScanOptions that's something like {file_path1: byte_range1, file_path2: byte_range2}. I think this would make sense.

Alternatively if this is not going to be supported, then this option ideally should only make sense for a dataset with one fragment. Perhaps I'll just add a check in the FragmentsToBatches function or something.

[asfimport]

Antoine Pitrou / @pitrou:
Ok, so perhaps byte ranges should actually be provided by the partitioning scheme ? Most partitioning schemes would only emit trivial byte ranges (i.e. the entire file), but the Substrait partitioning would emit whatever is in the Substrait plan.

Of course, it's up to the Substrait producer to ensure that these byte ranges actually make sense for the given data format. We shouldn't ask the file format implementations (for example CSV, Parquet or Orc) to accept innacurate byte ranges.