Method read_record_batch

arrow-ipc/src/reader.rs:548–621

Read the record batch, consuming the reader

(mut self)

    /// Read the record batch, consuming the reader
    fn read_record_batch(mut self) -> Result<RecordBatch, ArrowError> {
        let mut variadic_counts: VecDeque<i64> = self
            .batch
            .variadicBufferCounts()
            .into_iter()
            .flatten()
            .collect();

        let options = RecordBatchOptions::new().with_row_count(Some(self.batch.length() as usize));

        let schema = Arc::clone(&self.schema);
        if let Some(projection) = self.projection {
            let mut arrays = vec![];
            // project fields
            for (idx, field) in schema.fields().iter().enumerate() {
                // A projected field can appear more than once, so collect all matching positions.
                let mut child = None;
                for (proj_idx, projected_idx) in projection.iter().enumerate() {
                    if *projected_idx == idx {
                        if child.is_none() {
                            child = Some(self.create_array(field, &mut variadic_counts)?);
                        }

                        // Reuse the decoded array for duplicate projection entries.
                        arrays.push((proj_idx, child.as_ref().unwrap().clone()));
                    }
                }

                if child.is_none() {
                    self.skip_field(field, &mut variadic_counts)?;
                }
            }

            arrays.sort_by_key(|t| t.0);

            let schema = Arc::new(schema.project(projection)?);
            let columns = arrays.into_iter().map(|t| t.1).collect::<Vec<_>>();

            if self.skip_validation.get() {
                // Safety: setting `skip_validation` requires `unsafe`, user assures data is valid
                unsafe {
                    Ok(RecordBatch::new_unchecked(
                        schema,
                        columns,
                        self.batch.length() as usize,
                    ))
                }
            } else {
                assert!(variadic_counts.is_empty());
                RecordBatch::try_new_with_options(schema, columns, &options)
            }
        } else {
            let mut children = vec![];
            // keep track of index as lists require more than one node
            for field in schema.fields() {
                let child = self.create_array(field, &mut variadic_counts)?;
                children.push(child);
            }