Configurable coalesce ranges

[kylebarron]

By default, object_store has a hard-coded coalesce range
https://github.com/apache/arrow-rs/blob/a0c3186c55ac8ed3f6b8a15d1305548fd6305ebb/object_store/src/lib.rs#L643-L650

However, coalesce_ranges is public, so we can allow users to override how to coalesce ranges. Ref #5

[feefladder]

object_store is quite a large dependency. I would pull in coalesce_ranges (or ask object_store for a stub crate) and then be able to drop the object_store dependency.

[kylebarron]

We can make object_store (and reqwest) optional dependencies.

But we don't need to implement or vendor coalesce_ranges ourselves. The nice thing about having our AsyncFileReader trait is that the Rust user can decide how to implement get_byte_ranges, and they can implement range merging in their own code.

[feefladder]

hmmm.. I may whip up some async-cog crate thing then that handles those things.... Will PR for optional deps then

[kylebarron]

We can and should define some of those basic middlewares in this library, and then users can decide which they want to use.

[feefladder]

...so how could we implement a middleware that does coalesce_ranges without pulling its functionality in-tree? Would a user then do something like:

struct MyReader {
    href: String
}

#[async_trait] // if not wasm
impl AsyncFileReader for MyReader {
  fn get_bytes(&self, range: Range<u64>) -> AsyncTiffResult<Bytes> {
    let mut req = ehttp::Request::get(self.href);
    req.headers.insert("Range", write!("bytes={}-{}",range.start,range.end+1)
    Ok(Bytes::copy_from_slice(ehttp::fetch_async(req).await.unwrap().bytes) // no sanitizing
  }
}

type EhttpCogReader = PrefetchReader<CoalesceReader<MyReader>>

pub struct CoalesceReader {
  inner: Arc<dyn AsyncFileReader>
}

#[async_trait] // if not wasm
impl AsyncFileReader for CoalesceReader {
  fn get_bytes(&self, range: Range<u64>) -> AsyncTiffResult<Bytes> {
    self.inner.get_bytes(range).await
  }

  fn get_byte_ranges(&self, ranges: Vec<Range<u64>>) -> AsyncTiffResult<Vec<Bytes>> {
    coalesce_ranges(ranges, |range| self.get_bytes(range), 1024*1024)
  }
}