[C++] Implement a read range process without caching

[asfimport]

The current ReadRangeCache is mixing caching with coalescing and making difficult to implement readers capable to really perform concurrent reads on coalesced data (see this github comment for additional context); for instance, right now the prebuffering feature of those readers cannot handle concurrent invocations.

The goal for this ticket is to implement a similar component to ReadRangeCache for performing non-cache reads (doing only the coalescing part instead).  So, once we have that new capability, we can port the parquet and IPC readers to this new component and keep improving the reading process (that would be part of other set of follow-up tickets).  Similar ideas were mentioned here #32846

Maybe a good place to implement this new capability is inside the file system abstraction (as part of a dedicated method to read coalesced data) and where the abstract file system can provide a default implementation.

Reporter: Percy Camilo Triveño Aucahuasi / @aucahuasi
Assignee: Percy Camilo Triveño Aucahuasi / @aucahuasi

Related issues:

• feather.read_table 150x slower when reading columns in newer versions (relates to)
• [C++] ReadRangeCache should not retain data after read (relates to)
• [C++] Add opaque device id identification to InputStream (relates to)

PRs and other links:

• GitHub Pull Request #14723
• GitHub Pull Request #14747

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

[asfimport]

David Li / @lidavidm:
I'd like to raise my comments in ARROW-17913 and ARROW-17917 as well, especially https://issues.apache.org/jira/browse/ARROW-17913?focusedCommentId=17614155&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-17614155 . Note Weston offers suggestions there too about how we might handle this.

The API would probably be an extension of RandomAccessFile::ReadAsync. The file system would come into play by returning a subclass of RAF that overrides the new method and does coalescing appropriate to the underlying device

[asfimport]

David Li / @lidavidm:
(Of course, that's only if you choose to go with an explicit API, vs Weston's suggestion of possibly doing it 'automagically')

[asfimport]

Weston Pace / @westonpace:
On reflection, I don't really prefer my automagic suggestion. I think an explicit multi-read API added to the filesystem would be a good way to go. I don't see it as an extension of ReadAsync though. Something like:

  /// \brief Request multiple reads at once
  ///
  /// The underlying filesystem may optimize these reads by coalescing small reads into
  /// large reads or by breaking up large reads into multiple parallel smaller reads.  The
  /// reads should be issued in parallel if it makes sense for the filesystem.
  ///
  /// One future will be returned for each input read range.  Multiple returned futures
  /// may correspond to a single read.  Or, a single returned future may be a combined
  /// result of several individual reads.
  ///
  /// \param[in] ranges The ranges to read
  /// \return A future that will complete with the data from the requested range is
  /// available
  virtual std::vector<Future<std::shared_ptr<Buffer>>> ReadManyAsync(
      const IOContext&, const std::vector<ReadRange>& ranges);

There could be a default implementation (perhaps relying on configurable protected min_hole_size_ and max_contiguous_read_size_ variables) so that filesystems would only need to provide a specialized alternative where it made sense. In the future it would be interesting to benchmark and see if preadv can be used to provide a more optimized version for the local filesystem.

I'd also be curious to know how an API like this could be adapted (or whether my proposal fits) for something like io_uring @save-buffer

[asfimport]

David Li / @lidavidm:
Just to be clear: to the filesystem, or on the reader itself?

Also, I'm not clear on: "Multiple returned futures may correspond to a single read. Or, a single returned future may be a combined result of several individual reads." Isn't this saying the same thing twice?

[asfimport]

Weston Pace / @westonpace:

Just to be clear: to the filesystem, or on the reader itself?

Oops, I mean on RandomAccessFile.

Also, I'm not clear on: "Multiple returned futures may correspond to a single read. Or, a single returned future may be a combined result of several individual reads." Isn't this saying the same thing twice?

I might call

file->ReadMany({0, 3}, {3, 8}, {1024, 16Mi})
```.

The filesystem could then implement this as:

std::vector futures;

The first two futures correspond to the same read

Future coalesced_read = ReadAsync(0, 8);
futures.push_back(coalesced_read.Then(buf => buf.Split(0, 3)));
futures.push_back(coalesced_read.Then(buf => buf.Split(3, 5)));

The third future corresponds to two reads

Future part_one = ReadAsync(1024, 8Mi);
Future part_two = ReadAsync(1024+8Mi, 8Mi-1024);
futures.push_back(AllComplete({part_one, part_two}).Then(bufs => Concatenate(bufs));

[asfimport]

David Li / @lidavidm:
Ah ok I was forgetting that we could theoretically split up reads. Thanks!

[asfimport]

Sasha Krassovsky / @save-buffer:
Linking io_uring to the Future API could be a bit challenging because Linux doesn't have a builtin thread pool API, so you still end up having to poll the completion queue. I was thinking this could probably be integrated into the Executor implementation we have. In between tasks it would poll the completion queue and schedule the individual continuation futures that get finished. Using preadv would map pretty well but it would only ever return one future, I guess something like Future<std::vector<Buffer>>.

 

 

[asfimport]

Weston Pace / @westonpace:

Linking io_uring to the Future API could be a bit challenging because Linux doesn't have a builtin thread pool API, so you still end up having to poll the completion queue. I was thinking this could probably be integrated into the Executor implementation we have. In between tasks it would poll the completion queue and schedule the individual continuation futures that get finished.

Does io_uring have any kind of select-style API method like "wait until at least one pending read has finished"

Using preadv would map pretty well but it would only ever return one future, I guess something like Future<std::vector<Buffer>>.

Maybe this would be a better interface. I suppose it would boil down on whether or not the code can proceed with partial results. I think there might be cases where this is true but I'd really have to dive into the readers to recall exactly.

[asfimport]

David Li / @lidavidm:
For Parquet, the reader can submit reads for all ColumnChunks at once, and then decode them as they come back. So a single Future may not be great here? (Especially if the chunks are very large and don't get coalesced.)

[asfimport]

Sasha Krassovsky / @save-buffer:
Yes, io_uring has io_uring_wait_cqe which will wait until at least one response has finished. You can also do io_uring_wait_cqe_nr I think which will wait until a specific number of tasks have finished. 

 

We could still use io_uring without preadv. We'd just have to ensure that the ring is big enough to handle as many requests as we need. 

[asfimport]

Percy Camilo Triveño Aucahuasi / @aucahuasi:
Thanks for all the feedback!

Quick question: I see we have {}arrow::io::CacheOptions{}, but it seems that most of its contents are related to coalescing (except perhaps {}arrow::io::CacheOptions::lazy{}); does it sounds a good idea to rename it to arrow::io::CoalesceOptions and move it into {}interfaces.h{}?

The reason why I think we may need to move it into interfaces.h is that ReadManyAsync would be part of RandomAccessFile and we may need some of the attributes we have in arrow::io::CacheOptions (like hole_size_limit and {}range_size_limit{}) ... mostly becouse for the the implementation of ReadManyAsync we may need to use arrow::io::internal::CoalesceReadRanges and this function needs hole_size_limit and {}range_size_limit{}.

[asfimport]

David Li / @lidavidm:
Sounds reasonable to me, at first glance. Maybe leave behind a using CacheOptions = CoalesceOptions for compatibility (can you deprecate a using declaration?)

[asfimport]

Weston Pace / @westonpace:
I would think that coalescing would be an implementation detail specific to individual filesystems. For example, a local filesystem might not bother with it (the OS already does coalescing). However, an S3 filesystem would definitely want to coalesce. So if we add a ReadManyAsync then I think it should not have a cache options parameter. Instead that should be a property of the filesystem if it needs to be configurable.

[asfimport]

Percy Camilo Triveño Aucahuasi / @aucahuasi:

Sounds reasonable to me, at first glance. Maybe leave behind a using CacheOptions = CoalesceOptions for compatibility (can you deprecate a using declaration?)
David, I think we can deprecate it (in case we choose to rename&move it).
So if we add a ReadManyAsync then I think it should not have a cache options parameter. Instead that should be a property of the filesystem if it needs to be configurable.
Weston, yes initially I thought that CoalesceOptions would be part of arrow::io::IOContext (as an attribute) and ReadManyAsync could use/pass the CoalesceOptions to the filesystem.  But it make sense to let the filesystem handle all of that, so in that case:

1. we still may choose to rename arrow::io::CacheOptions to {}arrow::io::{}{}CoalesceOptions{} and move it into {}interfaces.h{}, so each filesystem's ctor will require {}arrow::io::CoalesceOptions{}.

2. or we just can include caching.h in every filesystem declaration without changing/renaming arrow::io::CacheOptions (so each filesystem's ctor will require {}arrow::io::{}{}CacheOptions{})

   Let me know which one sounds better to you, thanks.

[asfimport]

David Li / @lidavidm:
Ah sorry, ignore what I said - Weston is right. Though if you still want to rename it that sounds OK as long as we can have some sort of alias for compatibility.