/// Convert a stream of RecordBatches and a cmd to a stream of flights
Convert a stream of RecordBatches and a cmd to a stream of flights
Convert a stream of RecordBatches and a cmd to a stream of flights
pub async fn batch_stream_to_flight(
    cmd: &str,
    batches: SendableRecordBatchStream,
) -> Result<impl Stream<Item = Result<FlightData, Status>> + Send + Sync, Box<dyn Error>>
{
    let (sender, result) =
        tokio::sync::mpsc::unbounded_channel::<Result<FlightData, Status>>();
    let options = Arc::new(IpcWriteOptions::default());
    let mut flight_schema = flight_data_from_arrow_schema(&batches.schema(), &options);
    flight_schema.flight_descriptor = cmd_to_descriptor(&cmd);
    sender.send(Ok(flight_schema))?;
    // use channels to make stream sync (required by tonic)
    // TODO what happens with errors (currently all unwrapped in spawned task)
    tokio::spawn(async move {
        // then stream the rest
        batches
            .for_each(|batch_res| async {
                match batch_res {
                    Ok(batch) => {
                        let (dicts, data) =
                            flight_data_from_arrow_batch(&batch, &options);
                        dicts
                            .into_iter()
                            .chain(std::iter::once(data))
                            .for_each(|flight| (&sender).send(Ok(flight)).unwrap());
                    }
                    Err(err) => {
                        (&sender)
                            .send(Err(Status::aborted(format!("{}", err))))
                            .unwrap();
                    }
                };
            })
            .await;
    });
    Ok(UnboundedReceiverStream::new(result))
}
pub async fn batch_stream_to_flight ( cmd : & str , batches : SendableRecordBatchStream , ) -> Result < impl Stream < Item = Result < FlightData , Status > > + Send + Sync , Box < dyn Error > > { let ( sender , result ) = tokio :: sync :: mpsc :: unbounded_channel :: < Result < FlightData , Status > > ( ) ; let options = Arc :: new ( IpcWriteOptions :: default ( ) ) ; let mut flight_schema = flight_data_from_arrow_schema ( & batches . schema ( ) , & options ) ; flight_schema . flight_descriptor = cmd_to_descriptor ( & cmd ) ; sender . send ( Ok ( flight_schema ) ) ? ; // use channels to make stream sync (required by tonic) // TODO what happens with errors (currently all unwrapped in spawned task) tokio :: spawn ( async move { // then stream the rest batches . for_each ( | batch_res | async { match batch_res { Ok ( batch ) => { let ( dicts , data ) = flight_data_from_arrow_batch ( & batch , & options ) ; dicts . into_iter ( ) . chain ( std :: iter :: once ( data ) ) . for_each ( | flight | ( & sender ) . send ( Ok ( flight ) ) . unwrap ( ) ) ; } Err ( err ) => { ( & sender ) . send ( Err ( Status :: aborted ( format ! ( " " err ) ) ) ) . unwrap ( ) ; } } ; } ) . await ; } ) ; Ok ( UnboundedReceiverStream :: new ( result ) ) }
Convert a stream of RecordBatches and a cmd to a stream of flights
Convert a stream of RecordBatches and a cmd to a stream of flights
cmd
&str
batches
SendableRecordBatchStream