/// Convert a flight stream to a tuple with the cmd in the first flight and a stream of RecordBatch
Convert a flight stream to a tuple with the cmd in the first flight and a stream of RecordBatch
Convert a flight stream to a tuple with the cmd in the first flight and a stream of RecordBatch
pub async fn flight_to_batches(
    flights: tonic::Streaming<FlightData>,
) -> Result<(String, impl Stream<Item = ArrowResult<RecordBatch>>), Box<dyn Error>> {
    let mut flights = Box::pin(flights);
    let flight_data = flights.next().await.unwrap()?;
    let schema = Arc::new(Schema::try_from(&flight_data)?);
    let cmd = descriptor_to_cmd(flight_data.flight_descriptor)?;
    // all the remaining stream messages should be dictionary and record batches
    let record_batch_stream = flights.map(move |flight_data_res| match flight_data_res {
        Ok(flight_data) => {
            flight_data_to_arrow_batch(&flight_data, Arc::clone(&schema), &[])
        }
        Err(e) => Err(ArrowError::ExternalError(Box::new(e))),
    });
    Ok((cmd, record_batch_stream))
}
pub async fn flight_to_batches ( flights : tonic :: Streaming < FlightData > , ) -> Result < ( String , impl Stream < Item = ArrowResult < RecordBatch > > ) , Box < dyn Error > > { let mut flights = Box :: pin ( flights ) ; let flight_data = flights . next ( ) . await . unwrap ( ) ? ; let schema = Arc :: new ( Schema :: try_from ( & flight_data ) ? ) ; let cmd = descriptor_to_cmd ( flight_data . flight_descriptor ) ? ; // all the remaining stream messages should be dictionary and record batches let record_batch_stream = flights . map ( move | flight_data_res | match flight_data_res { Ok ( flight_data ) => { flight_data_to_arrow_batch ( & flight_data , Arc :: clone ( & schema ) , & [ ] ) } Err ( e ) => Err ( ArrowError :: ExternalError ( Box :: new ( e ) ) ) , } ) ; Ok ( ( cmd , record_batch_stream ) ) }
Convert a flight stream to a tuple with the cmd in the first flight and a stream of RecordBatch
Convert a flight stream to a tuple with the cmd in the first flight and a stream of RecordBatch
flights
tonic::Streaming<FlightData>