Function get_dictionary_values

arrow-ipc/src/reader.rs:867–912

Given a dictionary batch IPC message/body along with the full state of a stream including schema, dictionary cache, metadata, and other flags, this function will parse the buffer into an array of dictionary values.

(
    buf: &Buffer,
    batch: crate::DictionaryBatch,
    schema: &Schema,
    dictionaries_by_id: &mut HashMap<i64, ArrayRef>,
    metadata: &MetadataVersion,
    require_alignment: bool,
    skip_v

/// stream including schema, dictionary cache, metadata, and other flags, this
/// function will parse the buffer into an array of dictionary values.
fn get_dictionary_values(
    buf: &Buffer,
    batch: crate::DictionaryBatch,
    schema: &Schema,
    dictionaries_by_id: &mut HashMap<i64, ArrayRef>,
    metadata: &MetadataVersion,
    require_alignment: bool,
    skip_validation: UnsafeFlag,
) -> Result<ArrayRef, ArrowError> {
    let id = batch.id();
    #[allow(deprecated)]
    let fields_using_this_dictionary = schema.fields_with_dict_id(id);
    let first_field = fields_using_this_dictionary.first().ok_or_else(|| {
        ArrowError::InvalidArgumentError(format!("dictionary id {id} not found in schema"))
    })?;

    // As the dictionary batch does not contain the type of the
    // values array, we need to retrieve this from the schema.
    // Get an array representing this dictionary's values.
    let dictionary_values: ArrayRef = match first_field.data_type() {
        DataType::Dictionary(_, value_type) => {
            // Make a fake schema for the dictionary batch.
            let value = value_type.as_ref().clone();
            let schema = Schema::new(vec![Field::new("", value, true)]);
            // Read a single column
            let record_batch = RecordBatchDecoder::try_new(
                buf,
                batch.data().unwrap(),
                Arc::new(schema),
                dictionaries_by_id,
                metadata,
            )?
            .with_require_alignment(require_alignment)
            .with_skip_validation(skip_validation)
            .read_record_batch()?;

            Some(record_batch.column(0).clone())
        }
        _ => None,
    }
    .ok_or_else(|| {
        ArrowError::InvalidArgumentError(format!("dictionary id {id} not found in schema"))
    })?;

    Ok(dictionary_values)
}

/// Read the data for a given block
fn read_block<R: Read + Seek>(mut reader: R, block: &Block) -> Result<Buffer, ArrowError> {