| 326 | } |
| 327 | |
| 328 | async fn execute_flight( |
| 329 | client: &mut FlightSqlServiceClient<Channel>, |
| 330 | info: FlightInfo, |
| 331 | ) -> Result<Vec<RecordBatch>> { |
| 332 | let schema = Arc::new(Schema::try_from(info.clone()).context("valid schema")?); |
| 333 | let mut batches = Vec::with_capacity(info.endpoint.len() + 1); |
| 334 | batches.push(RecordBatch::new_empty(schema)); |
| 335 | info!("decoded schema"); |
| 336 | |
| 337 | for endpoint in info.endpoint { |
| 338 | let Some(ticket) = &endpoint.ticket else { |
| 339 | bail!("did not get ticket"); |
| 340 | }; |
| 341 | |
| 342 | let mut flight_data = client.do_get(ticket.clone()).await.context("do get")?; |
| 343 | log_metadata(flight_data.headers(), "header"); |
| 344 | |
| 345 | let mut endpoint_batches: Vec<_> = (&mut flight_data) |
| 346 | .try_collect() |
| 347 | .await |
| 348 | .context("collect data stream")?; |
| 349 | batches.append(&mut endpoint_batches); |
| 350 | |
| 351 | if let Some(trailers) = flight_data.trailers() { |
| 352 | log_metadata(&trailers, "trailer"); |
| 353 | } |
| 354 | } |
| 355 | info!("received data"); |
| 356 | |
| 357 | Ok(batches) |
| 358 | } |
| 359 | |
| 360 | fn construct_record_batch_from_params( |
| 361 | params: &[(String, String)], |