Method execute_ingest

arrow-flight/src/sql/client.rs:231–264

Execute a bulk ingest on the server and return the number of records added

(
        &mut self,
        command: CommandStatementIngest,
        stream: S,
    )

    /// Execute a bulk ingest on the server and return the number of records added
    pub async fn execute_ingest<S>(
        &mut self,
        command: CommandStatementIngest,
        stream: S,
    ) -> Result<i64>
    where
        S: Stream<Item = crate::error::Result<RecordBatch>> + Send + 'static,
    {
        let (sender, receiver) = futures::channel::oneshot::channel();

        let descriptor = FlightDescriptor::new_cmd(command.as_any().encode_to_vec());
        let flight_data = FlightDataEncoderBuilder::new()
            .with_flight_descriptor(Some(descriptor))
            .build(stream);

        // Intercept client errors and send them to the one shot channel above
        let flight_data = Box::pin(flight_data);
        let flight_data: FallibleRequestStream<FlightData, FlightError> =
            FallibleRequestStream::new(sender, flight_data);

        let req = self.set_request_headers(flight_data.into_streaming_request())?;
        let mut result = self.flight_client.do_put(req).await?.into_inner();

        // check if the there were any errors in the input stream provided note
        // if receiver.await fails, it means the sender was dropped and there is
        // no message to return.
        if let Ok(msg) = receiver.await {
            return Err(FlightError::ExternalError(Box::new(msg)));
        }

        let result = result.message().await?.unwrap();
        let result: DoPutUpdateResult = Message::decode(&*result.app_metadata)?;
        Ok(result.record_count)
    }

    /// Request a list of catalogs as tabular FlightInfo results
    pub async fn get_catalogs(&mut self) -> Result<FlightInfo> {