[POC][wip] improve json reading with concurrent requests

[zachschuermann]

related: #595

note this is out of order due to spawn - instead I think i'll just do futures combinators

[scovich]

I may have badly understood how this async stuff works, but I left some comments anyway :P

[scovich]

I didn't think traits were allowed to provide associated types like this in stable rust... did something stabilize recently?

[scovich]

Also, does this work?

Suggested change

	type JoinHandle<T>: Future<Output = Result<T, Self::JoinError>> + Send
	where
	T: Send;
	type JoinHandle<T: Send>: Future<Output = Result<T, Self::JoinError>> + Send;

[scovich]

grammar nit

Suggested change

	// Note: we throw away the JoinHandle since we don't need to wait for it's return
	// Note: we throw away the JoinHandle since we don't need to wait for its return

[scovich]

I don't think this would compile with that semicolon blocking the now-required return value?

[scovich]

Double checking -- the only error recv ever returns is HUP, and so converting that to an iterator None result is the desired behavior?

Suggested change

	self.rx
	.recv()
	.ok()
	.map(|r| r.map(|rb| Box::new(ArrowEngineData::new(rb)) as Box<dyn EngineData>))
	let r = self.rx.recv().ok()?; // HUP -> iterator finished
	r.map(|rb| Box::new(ArrowEngineData::new(rb)) as _)

[scovich]

Suggested change

	for file in files.into_iter() {
	for file in files {

https://doc.rust-lang.org/std/iter/index.html#for-loops-and-intoiterator

[scovich]

Why do all tasks need to complete before we can return the iterator?
(prob need to clone the file list and task executor arc before launching tasks tho -- else will have borrow lifetime issues)

[scovich]

I'm not seeing how this preserves order as required by log replay? It looks like the various tasks all race to append their stream chunks to the queue?

[scovich]

If I understand correctly, this call causes the task to start running. Thus launching an open call for each file. But does the open make progress without somebody waiting on it, so that the first stream.next() call gets triggered? I would hope so.

Meanwhile, given the batch ordering issues (see comment below), it seems like we should create an iterator of handles, where each handle's return type is a DeltaResult<BoxStream> that the ChunkIter can then await directly?

let streams = files.iter().cloned().map(|f| {
      ...
    runtime.spawn(async move {
        Ok(f.open(...)?.await?)
    })
});

and then, assuming the iterator wraps a peekable version of streams and also has a cloned arc of the task executor:

fn next(&mut self) -> Option<Self::Item> {
    self.task_executor.block_on(async {
        while let Some(stream) = self.streams.peek() {
            match stream {
                Ok(stream) => {
                    if let Some(result) = stream.next().await {
                        return Some(result);
                    }
                    let _ = streams.pop(); // current stream is exhausted
                }
                Err(err) => {
                    let _ = streams.pop(); // current stream failed
                    return Some(Err(err))
                }
            };
        }
        None // all streams exhausted
    }
}