Convert `FlightData` (with supplied schema and dictionaries) to an arrow `RecordBatch`.
(
data: &FlightData,
schema: SchemaRef,
dictionaries_by_id: &HashMap<i64, ArrayRef>,
)
| 53 | |
| 54 | /// Convert `FlightData` (with supplied schema and dictionaries) to an arrow `RecordBatch`. |
| 55 | pub fn flight_data_to_arrow_batch( |
| 56 | data: &FlightData, |
| 57 | schema: SchemaRef, |
| 58 | dictionaries_by_id: &HashMap<i64, ArrayRef>, |
| 59 | ) -> Result<RecordBatch, ArrowError> { |
| 60 | // check that the data_header is a record batch message |
| 61 | let message = arrow_ipc::root_as_message(&data.data_header[..]) |
| 62 | .map_err(|err| ArrowError::ParseError(format!("Unable to get root as message: {err:?}")))?; |
| 63 | |
| 64 | message |
| 65 | .header_as_record_batch() |
| 66 | .ok_or_else(|| { |
| 67 | ArrowError::ParseError( |
| 68 | "Unable to convert flight data header to a record batch".to_string(), |
| 69 | ) |
| 70 | }) |
| 71 | .map(|batch| { |
| 72 | reader::read_record_batch( |
| 73 | &Buffer::from(data.data_body.as_ref()), |
| 74 | batch, |
| 75 | schema, |
| 76 | dictionaries_by_id, |
| 77 | None, |
| 78 | &message.version(), |
| 79 | ) |
| 80 | })? |
| 81 | } |
| 82 | |
| 83 | /// Convert `RecordBatch`es to wire protocol `FlightData`s |
| 84 | pub fn batches_to_flight_data( |