/// Ensure that the record batch has the given schema.
///
/// This does multiple things:
///
/// 1. Dictionary type recovery
/// 2. NULL-column creation
///
/// # Dictionary Type Recovery
///
/// Cast arrays in record batch to be the type of schema. This is a workaround for
/// <https://github.com/influxdata/influxdb_iox/pull/4273> where the Flight API doesn't necessarily
/// return the same schema as was provided by the ingester.
///
/// Namely, dictionary encoded columns (e.g. tags) are returned as `DataType::Utf8` even when they
/// were sent as `DataType::Dictionary(Int32, Utf8)`.
///
/// # NULL-column Creation
///
/// If a column is absent in an ingester partition it will not be part of record batch even when
/// the querier requests it. In that case we create it as "all NULL" column with the appropriate
/// type.
///
/// An alternative would be to remove the column from the schema of the appropriate
/// [`IngesterPartition`]. However, since a partition may contain multiple record batches and we do
/// not want to assume that the presence/absence of columns is identical for all of them, we fix
/// this here.
Ensure that the record batch has the given schema.
This does multiple things.

1. Dictionary type recovery
2. NULL-column creation
Ensure that the record batch has the given schema . This does multiple things . 1 . Dictionary type recovery 2 . NULL - column creation
fn ensure_schema(batch: RecordBatch, expected_schema: &Schema) -> Result<RecordBatch> {
    let actual_schema = batch.schema();
    let desired_fields = expected_schema.iter().map(|(_, f)| f);
    let new_columns = desired_fields
        .map(|desired_field| {
            let desired_type = desired_field.data_type();
            // find column by name
            match actual_schema.column_with_name(desired_field.name()) {
                Some((idx, _field)) => {
                    let col = batch.column(idx);
                    let actual_type = col.data_type();
                    // check type
                    if desired_type != actual_type {
                        if let DataType::Dictionary(_key_type, value_type) = desired_type.clone() {
                            if value_type.as_ref() == actual_type {
                                // convert
                                return arrow::compute::cast(col, desired_type).context(
                                    ConvertingRecordBatchSnafu {
                                        column_name: desired_field.name(),
                                        data_type: desired_type.clone(),
                                    },
                                );
                            }
                        }
                        RecordBatchTypeSnafu {
                            column_name: desired_field.name(),
                            actual_data_type: actual_type.clone(),
                            desired_data_type: desired_type.clone(),
                        }
                        .fail()
                    } else {
                        Ok(Arc::clone(col))
                    }
                }
                None => Ok(arrow::array::new_null_array(desired_type, batch.num_rows())),
            }
        })
        .collect::<Result<Vec<_>>>()?;
    RecordBatch::try_new(expected_schema.as_arrow(), new_columns).context(CreatingRecordBatchSnafu)
}
fn ensure_schema ( batch : RecordBatch , expected_schema : & Schema ) -> Result < RecordBatch > { let actual_schema = batch . schema ( ) ; let desired_fields = expected_schema . iter ( ) . map ( | ( _ , f ) | f ) ; let new_columns = desired_fields . map ( | desired_field | { let desired_type = desired_field . data_type ( ) ; // find column by name match actual_schema . column_with_name ( desired_field . name ( ) ) { Some ( ( idx , _field ) ) => { let col = batch . column ( idx ) ; let actual_type = col . data_type ( ) ; // check type if desired_type != actual_type { if let DataType :: Dictionary ( _key_type , value_type ) = desired_type . clone ( ) { if value_type . as_ref ( ) == actual_type { // convert return arrow :: compute :: cast ( col , desired_type ) . context ( ConvertingRecordBatchSnafu { column_name : desired_field . name ( ) , data_type : desired_type . clone ( ) , } , ) ; } } RecordBatchTypeSnafu { column_name : desired_field . name ( ) , actual_data_type : actual_type . clone ( ) , desired_data_type : desired_type . clone ( ) , } . fail ( ) } else { Ok ( Arc :: clone ( col ) ) } } None => Ok ( arrow :: array :: new_null_array ( desired_type , batch . num_rows ( ) ) ) , } } ) . collect :: < Result < Vec < _ > > > ( ) ? ; RecordBatch :: try_new ( expected_schema . as_arrow ( ) , new_columns ) . context ( CreatingRecordBatchSnafu ) }
Ensure that the record batch has the given schema.
Ensure that the record batch has the given schema .
batch
RecordBatch
expected_schema
&Schema
dictionary type recovery
Cast arrays in record batch to be the type of schema. This is a workaround for
where the Flight API doesn't necessarily
return the same schema as was provided by the ingester.

Namely, dictionary encoded columns  are returned as `DataType::Utf8` even when they
were sent as `DataType::Dictionary(Int32, Utf8)`.
Cast arrays in record batch to be the type of schema . This is a workaround for where the Flight API doesn ' t necessarily return the same schema as was provided by the ingester . Namely dictionary encoded columns are returned as ` DataType :: Utf8 ` even when they were sent as ` DataType :: Dictionary ( Int32 Utf8 ) ` .
null-column creation
If a column is absent in an ingester partition it will not be part of record batch even when
the querier requests it. In that case we create it as "all NULL" column with the appropriate
type.

An alternative would be to remove the column from the schema of the appropriate
[`IngesterPartition`]. However, since a partition may contain multiple record batches and we do
not want to assume that the presence/absence of columns is identical for all of them, we fix
this here.
If a column is absent in an ingester partition it will not be part of record batch even when the querier requests it . In that case we create it as " all NULL " column with the appropriate type . An alternative would be to remove the column from the schema of the appropriate [ ` IngesterPartition ` ] . However since a partition may contain multiple record batches and we do not want to assume that the presence / absence of columns is identical for all of them we fix this here .