async fn download_and_scan_parquet(
    projection: Vec<usize>,
    path: object_store::path::Path,
    object_store: Arc<DynObjectStore>,
    tx: tokio::sync::mpsc::Sender<ArrowResult<RecordBatch>>,
) -> Result<(), ReadError> {
    let read_stream = object_store.get(&path).await?;

    let data = match read_stream {
        GetResult::File(mut f, _) => {
            trace!(?path, "Using file directly");
            let l = f.metadata().await?.len();
            let mut buf = Vec::with_capacity(l as usize);
            f.read_to_end(&mut buf).await?;
            buf
        }
        GetResult::Stream(read_stream) => {
            let chunks: Vec<_> = read_stream.try_collect().await?;

            let mut buf = Vec::with_capacity(chunks.iter().map(|c| c.len()).sum::<usize>());
            for c in chunks {
                buf.extend(c);
            }

            buf
        }
    };

    // Size of each batch
    let batch_size = 1024; // Todo: make a constant or policy for this

    let cursor = SliceableCursor::new(data);
    let file_reader = SerializedFileReader::new(cursor)?;
    let mut arrow_reader = ParquetFileArrowReader::new(Arc::new(file_reader));

    let mask = ProjectionMask::roots(
        arrow_reader.get_metadata().file_metadata().schema_descr(),
        projection,
    );
    let record_batch_reader = arrow_reader.get_record_reader_by_columns(mask, batch_size)?;

    for batch in record_batch_reader {
        if tx.send(batch).await.is_err() {
            debug!("Receiver hung up - exiting");
            break;
        }
    }

    debug!(?path, "Completed parquet download & scan");

    Ok(())
}
/// Downloads the specified parquet file to a local temporary file
/// and push the [`RecordBatch`] contents over `tx`, projecting the specified
/// column indexes.
///
/// This call MAY download a parquet file from object storage, temporarily
/// spilling it to disk while it is processed.
Downloads the specified parquet file to a local temporary file
and push the [`RecordBatch`] contents over `tx`, projecting the specified
column indexes.

This call MAY download a parquet file from object storage, temporarily
spilling it to disk while it is processed.
Downloads the specified parquet file to a local temporary file and push the [ ` RecordBatch ` ] contents over ` tx ` projecting the specified column indexes . This call MAY download a parquet file from object storage temporarily spilling it to disk while it is processed .
async fn download_and_scan_parquet(
    projection: Vec<usize>,
    path: object_store::path::Path,
    object_store: Arc<DynObjectStore>,
    tx: tokio::sync::mpsc::Sender<ArrowResult<RecordBatch>>,
) -> Result<(), ReadError> {
    let read_stream = object_store.get(&path).await?;
    let data = match read_stream {
        GetResult::File(mut f, _) => {
            trace!(?path, "Using file directly");
            let l = f.metadata().await?.len();
            let mut buf = Vec::with_capacity(l as usize);
            f.read_to_end(&mut buf).await?;
            buf
        }
        GetResult::Stream(read_stream) => {
            let chunks: Vec<_> = read_stream.try_collect().await?;
            let mut buf = Vec::with_capacity(chunks.iter().map(|c| c.len()).sum::<usize>());
            for c in chunks {
                buf.extend(c);
            }
            buf
        }
    };
    // Size of each batch
    let batch_size = 1024; // Todo: make a constant or policy for this
    let cursor = SliceableCursor::new(data);
    let file_reader = SerializedFileReader::new(cursor)?;
    let mut arrow_reader = ParquetFileArrowReader::new(Arc::new(file_reader));
    let mask = ProjectionMask::roots(
        arrow_reader.get_metadata().file_metadata().schema_descr(),
        projection,
    );
    let record_batch_reader = arrow_reader.get_record_reader_by_columns(mask, batch_size)?;
    for batch in record_batch_reader {
        if tx.send(batch).await.is_err() {
            debug!("Receiver hung up - exiting");
            break;
        }
    }
    debug!(?path, "Completed parquet download & scan");
    Ok(())
}
async fn download_and_scan_parquet ( projection : Vec < usize > , path : object_store :: path :: Path , object_store : Arc < DynObjectStore > , tx : tokio :: sync :: mpsc :: Sender < ArrowResult < RecordBatch > > , ) -> Result < ( ) , ReadError > { let read_stream = object_store . get ( & path ) . await ? ; let data = match read_stream { GetResult :: File ( mut f , _ ) => { trace ! ( path " " ) ; let l = f . metadata ( ) . await ? . len ( ) ; let mut buf = Vec :: with_capacity ( l as usize ) ; f . read_to_end ( & mut buf ) . await ? ; buf } GetResult :: Stream ( read_stream ) => { let chunks : Vec < _ > = read_stream . try_collect ( ) . await ? ; let mut buf = Vec :: with_capacity ( chunks . iter ( ) . map ( | c | c . len ( ) ) . sum :: < usize > ( ) ) ; for c in chunks { buf . extend ( c ) ; } buf } } ; // Size of each batch let batch_size = 1024 ; // Todo: make a constant or policy for this let cursor = SliceableCursor :: new ( data ) ; let file_reader = SerializedFileReader :: new ( cursor ) ? ; let mut arrow_reader = ParquetFileArrowReader :: new ( Arc :: new ( file_reader ) ) ; let mask = ProjectionMask :: roots ( arrow_reader . get_metadata ( ) . file_metadata ( ) . schema_descr ( ) , projection , ) ; let record_batch_reader = arrow_reader . get_record_reader_by_columns ( mask , batch_size ) ? ; for batch in record_batch_reader { if tx . send ( batch ) . await . is_err ( ) { debug ! ( " " ) ; break ; } } debug ! ( path " " ) ; Ok ( ( ) ) }
Downloads the specified parquet file to a local temporary file
and push the [`RecordBatch`] contents over `tx`, projecting the specified
column indexes.
Downloads the specified parquet file to a local temporary file and push the [ ` RecordBatch ` ] contents over ` tx ` projecting the specified column indexes .
projection
Vec<usize>
path
object_store::path::Path
object_store
Arc<DynObjectStore>
tx
tokio::sync::mpsc::Sender<ArrowResult<RecordBatch>>
projection
Vec<usize>
path
object_store::path::Path
object_store
Arc<DynObjectStore>
tx
tokio::sync::mpsc::Sender<ArrowResult<RecordBatch>>