ARROW-10046: [Rust] [DataFusion] Made RecordBatchReader implement Iterator

[jorgecarleitao]

This is a proposal to change how we programmatically iterate over record batches in arrow and datafusion.

Instead of

pub fn collect(
    it: Arc<Mutex<dyn RecordBatchReader + Send + Sync>>,
) -> Result<Vec<RecordBatch>> {
    let mut reader = it.lock().unwrap();
    let mut results: Vec<RecordBatch> = vec![];
    loop {
        match reader.next_batch() {
            Ok(Some(batch)) => {
                results.push(batch);
            }
            Ok(None) => {
                // end of result set
                return Ok(results);
            }
            Err(e) => return Err(ExecutionError::from(e)),
        }
    }
}

use

pub fn collect(
    it: Arc<Mutex<dyn RecordBatchReader + Send + Sync>>,
) -> Result<Vec<RecordBatch>> {
    it.lock()
        .unwrap()
        .into_iter()
        .collect::<ArrowResult<Vec<_>>>()
        .map_err(|e| ExecutionError::from(e))
}

I.e. via the Iterator trait.

This allow us to write more expressive code, as well as offer a well documented and popular API to our users (Iterator).

Finally, this change also opens the possibility to implement future::Stream, the async version of Iterator.

[github-actions]

https://issues.apache.org/jira/browse/ARROW-10046

[jorgecarleitao]

FYI, I am trying to summarize how DataFusion iterates over data, and I came up with this summary.

[alamb]

While this will be a breaking change for any code that currently uses RecordBatchReader I think it is a significant improvement and I vote (not that I am sure what my vote counts for :) ) that it is merged in.

If the API breakage is a concern (by removing RecordBatchReader::next_batch I have a suggestion, inline, that could potentially mitigate that)

[andygrove]

Very nice. Thanks @jorgecarleitao