Repository: jorgecarleitao/arrow2
Pull request #1089 (2022-06-20 21:18:09)

Removed `Buffer::into_mut` and `make_mut` functions
The function `get_mut` covers both cases, making it the entry point for `cow` semantics.


codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#1089](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (c0c793a) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/d1ab4efa5c895513681bacd5e20660be882433ac?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (d1ab4ef) will **increase** coverage by `0.00%`.
> The diff coverage is `73.55%`.

```diff
@@           Coverage Diff           @@
##             main    #1089   +/-   ##
=======================================
  Coverage   81.07%   81.08%           
=======================================
  Files         367      367           
  Lines       35309    35293   -16     
=======================================
- Hits        28628    28618   -10     
+ Misses       6681     6675    -6     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/array/specification.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L3NwZWNpZmljYXRpb24ucnM=) | `91.46% <ø> (-0.31%)` | :arrow_down: |
| [src/buffer/bytes.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2J1ZmZlci9ieXRlcy5ycw==) | `74.13% <ø> (ø)` | |
| [src/buffer/immutable.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2J1ZmZlci9pbW11dGFibGUucnM=) | `88.88% <ø> (+4.93%)` | :arrow_up: |
| [src/io/parquet/write/binary/basic.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvYmluYXJ5L2Jhc2ljLnJz) | `68.14% <0.00%> (ø)` | |
| [src/bitmap/mutable.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2JpdG1hcC9tdXRhYmxlLnJz) | `96.86% <40.00%> (-0.66%)` | :arrow_down: |
| [src/array/primitive/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L3ByaW1pdGl2ZS9tb2QucnM=) | `79.53% <54.54%> (ø)` | |
| [src/array/utf8/mutable.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L3V0ZjgvbXV0YWJsZS5ycw==) | `81.33% <56.00%> (-2.67%)` | :arrow_down: |
| [src/array/utf8/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L3V0ZjgvbW9kLnJz) | `82.74% <59.25%> (+3.35%)` | :arrow_up: |
| [src/bitmap/immutable.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2JpdG1hcC9pbW11dGFibGUucnM=) | `81.48% <90.00%> (-1.54%)` | :arrow_down: |
| [src/array/binary/mutable.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2JpbmFyeS9tdXRhYmxlLnJz) | `87.97% <100.00%> (ø)` | |
| ... and [21 more](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Last update [d1ab4ef...c0c793a](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1089?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).


Patch:
diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md
index cc9211ef99..d81e645583 100644
--- a/DEVELOPMENT.md
+++ b/DEVELOPMENT.md
@@ -4,6 +4,13 @@ This crate follows the standard for developing a Rust library via `cargo`.
 The CI is our "ground truth" over the state of the library. Check out the different parts of
 the CI to understand how to test the different parts of this library locally.
 
+## Git clone with submodules
+The crate comes with additional submodules to aid with testing, to ensure you have them if you plan on testing, using `--recurse-submodules` will clone the submodules alongside the repository.
+
+```bash
+git clone --recurse-submodules https://github.com/jorgecarleitao/arrow2
+```
+
 ## Testing
 
 The simplest way to test the crate is to run 
@@ -18,6 +25,41 @@ This runs the tests of the crate without features. To run all features, use
 cargo test --features full
 ```
 
+If you get warnings about parquet files not existing, you can generate the test files by using Python:
+
+```bash
+# Ubuntu: sudo apt install python3-pip python3-venv
+# Mac: brew install python3
+# Archlinux: sudo pacman -Syu python
+
+# Create a virtual environment for python, to keep dependencies contained
+python3 -m venv venv
+
+# Activate the virtual environment
+source venv/bin/activate
+
+# Make sure pip is up to date
+pip install pip --upgrade
+
+# Install pyarrow, version 6
+pip install pyarrow==6
+
+# Generate the parquet files (this might take some time, depending on your computer setup)
+python parquet_integration/write_parquet.py
+
+# Get out of venv, back to normal terminal
+deactivate
+```
+
+If you receive warnings about other files not found (IPC), ensure you have all submodules:
+```bash
+# If you didn't clone with `git clone --recurse-submodules https://github.com/jorgecarleitao/arrow2`
+git submodule update --init --recursive
+
+# Update to the latest submodules
+git submodule update --recursive --remote
+```
+
 during development of particular parts of the crate, it is usually faster
 to reduce the feature set - the tests are gated to only the relevant tests
 of that feature set. For example, if improving JSON, you can use
diff --git a/src/array/utf8/mutable.rs b/src/array/utf8/mutable.rs
index b80c73e212..12a81d5c34 100644
--- a/src/array/utf8/mutable.rs
+++ b/src/array/utf8/mutable.rs
@@ -66,10 +66,9 @@ impl<O: Offset> Default for MutableUtf8Array<O> {
 impl<O: Offset> MutableUtf8Array<O> {
     /// Initializes a new empty [`MutableUtf8Array`].
     pub fn new() -> Self {
-        let offsets = vec![O::default()];
         Self {
             data_type: Self::default_data_type(),
-            offsets,
+            offsets: vec![O::default()],
             values: Vec::<u8>::new(),
             validity: None,
         }
diff --git a/src/buffer/bytes.rs b/src/buffer/bytes.rs
index 3e33fc7ce4..de9b6b73b5 100644
--- a/src/buffer/bytes.rs
+++ b/src/buffer/bytes.rs
@@ -31,10 +31,8 @@ impl Debug for Deallocation {
 
 /// A continuous, fixed-size, immutable memory region that knows how to de-allocate itself.
 ///
-/// In the most common case, this buffer is allocated using [`allocate_aligned`](alloc::allocate_aligned)
-/// and deallocated accordingly [`free_aligned`](alloc::free_aligned).
-/// When the region is allocated by a foreign allocator, [Deallocation::Foreign], this calls the
-/// foreign deallocator to deallocate the region when it is no longer needed.
+/// In the most common case, this buffer is allocated using Rust's native allocator.
+/// However, it may also be allocated by a foreign allocator, [Deallocation::Foreign].
 pub struct Bytes<T: NativeType> {
     /// inner data
     data: MaybeForeign<T>,
diff --git a/src/buffer/immutable.rs b/src/buffer/immutable.rs
index c4e73ce33d..8bbe0115d2 100644
--- a/src/buffer/immutable.rs
+++ b/src/buffer/immutable.rs
@@ -1,37 +1,37 @@
-use either::Either;
 use std::{iter::FromIterator, sync::Arc, usize};
 
 use crate::types::NativeType;
 
 use super::bytes::Bytes;
 
-/// [`Buffer`] is a contiguous memory region that can be shared across thread boundaries.
+/// [`Buffer`] is a contiguous memory region of plain old data types
+/// that can be shared across thread boundaries.
 ///
-/// The easiest way to think about `Buffer<T>` is being equivalent to
+/// The easiest way to think about [`Buffer<T>`] is being equivalent to
 /// a `Arc<Vec<T>>`, with the following differences:
 /// * `T` must be [`NativeType`]
-/// * slicing the buffer is `O(1)`
+/// * slicing and cloning is `O(1)`.
 /// * it supports external allocated memory (via FFI)
 ///
-/// The easiest way to create one is to use its implementation of `From` a `Vec`.
+/// The easiest way to create one is to use its implementation of `From<Vec<T>>`.
 ///
 /// # Examples
 /// ```
 /// use arrow2::buffer::Buffer;
 ///
-/// let buffer: Buffer<u32> = vec![1, 2, 3].into();
+/// let mut buffer: Buffer<u32> = vec![1, 2, 3].into();
 /// assert_eq!(buffer.as_ref(), [1, 2, 3].as_ref());
 ///
 /// // it supports copy-on-write semantics (i.e. back to a `Vec`)
-/// let vec: Vec<u32> = buffer.into_mut().right().unwrap();
-/// assert_eq!(vec, vec![1, 2, 3]);
+/// let vec: &mut [u32] = buffer.get_mut().unwrap();
+/// assert_eq!(vec, &mut [1, 2, 3]);
 ///
 /// // cloning and slicing is `O(1)` (data is shared)
-/// let buffer: Buffer<u32> = vec![1, 2, 3].into();
+/// let mut buffer: Buffer<u32> = vec![1, 2, 3].into();
 /// let slice = buffer.clone().slice(1, 1);
 /// assert_eq!(slice.as_ref(), [2].as_ref());
-/// // no longer possible to get a vec since `slice` and `buffer` share data
-/// let same: Buffer<u32> = buffer.into_mut().left().unwrap();
+/// // but cloning forbids getting mut since `slice` and `buffer` now share data
+/// assert_eq!(buffer.get_mut(), None);
 /// ```
 #[derive(Clone, PartialEq)]
 pub struct Buffer<T: NativeType> {
@@ -100,7 +100,7 @@ impl<T: NativeType> Buffer<T> {
         }
     }
 
-    /// Returns a new [Buffer] that is a slice of this buffer starting at `offset`.
+    /// Returns a new [`Buffer`] that is a slice of this buffer starting at `offset`.
     /// Doing so allows the same memory region to be shared between buffers.
     /// # Panics
     /// Panics iff `offset` is larger than `len`.
@@ -114,7 +114,7 @@ impl<T: NativeType> Buffer<T> {
         unsafe { self.slice_unchecked(offset, length) }
     }
 
-    /// Returns a new [Buffer] that is a slice of this buffer starting at `offset`.
+    /// Returns a new [`Buffer`] that is a slice of this buffer starting at `offset`.
     /// Doing so allows the same memory region to be shared between buffers.
     /// # Safety
     /// The caller must ensure `offset + length <= self.len()`
@@ -137,10 +137,10 @@ impl<T: NativeType> Buffer<T> {
         self.offset
     }
 
-    /// Gets a mutable reference to its underlying [`Vec`], if it not being shared.
+    /// Returns a mutable reference to its underlying [`Vec`], if possible.
     ///
-    /// This operation returns a [`Vec`] iff this [`Buffer`]:
-    /// * is not an offsetted slice of another [`Buffer`]
+    /// This operation returns [`Some`] iff this [`Buffer`]:
+    /// * has not been sliced with an offset
     /// * has not been cloned (i.e. [`Arc`]`::get_mut` yields [`Some`])
     /// * has not been imported from the c data interface (FFI)
     pub fn get_mut(&mut self) -> Option<&mut Vec<T>> {
@@ -151,32 +151,6 @@ impl<T: NativeType> Buffer<T> {
         }
     }
 
-    /// Converts this [`Buffer`] to either a [`Buffer`] or a [`Vec`], returning itself if the conversion
-    /// is not possible
-    ///
-    /// This operation returns a [`Vec`] iff this [`Buffer`]:
-    /// * is not an offsetted slice of another [`Buffer`]
-    /// * has not been cloned (i.e. [`Arc`]`::get_mut` yields [`Some`])
-    /// * has not been imported from the c data interface (FFI)
-    pub fn into_mut(mut self) -> Either<Self, Vec<T>> {
-        if let Some(vec) = self.get_mut() {
-            Either::Right(std::mem::take(vec))
-        } else {
-            Either::Left(self)
-        }
-    }
-
-    /// Converts this [`Buffer`] to a [`Vec`], cloning the data if needed, also
-    /// known as clone-on-write semantics.
-    ///
-    /// This function is O(1) under the same conditions that [`Self::into_mut`] returns `Vec`.
-    pub fn make_mut(self) -> Vec<T> {
-        match self.into_mut() {
-            Either::Left(data) => data.as_ref().to_vec(),
-            Either::Right(data) => data,
-        }
-    }
-
     /// Get the strong count of underlying `Arc` data buffer.
     pub fn shared_count_strong(&self) -> usize {
         Arc::strong_count(&self.data)
diff --git a/src/compute/arity_assign.rs b/src/compute/arity_assign.rs
index 79d8f579b3..b7fbdeab3d 100644
--- a/src/compute/arity_assign.rs
+++ b/src/compute/arity_assign.rs
@@ -66,11 +66,9 @@ where
                         // alloc new region
                         &immutable & rhs
                     }
-                    Either::Right(mut mutable) => {
+                    Either::Right(mutable) => {
                         // mutate in place
-                        let mut mutable_ref = &mut mutable;
-                        mutable_ref &= rhs;
-                        mutable.into()
+                        (mutable & rhs).into()
                     }
                 }
             });
diff --git a/src/io/flight/mod.rs b/src/io/flight/mod.rs
index b439ced2d9..8aa4a1de0e 100644
--- a/src/io/flight/mod.rs
+++ b/src/io/flight/mod.rs
@@ -11,12 +11,14 @@ use crate::{
     error::{Error, Result},
     io::ipc::read,
     io::ipc::write,
-    io::ipc::write::common::{encode_chunk, DictionaryTracker, EncodedData, WriteOptions},
+    io::ipc::write::common::{encode_chunk, DictionaryTracker, EncodedData},
 };
 
-pub use super::ipc::write::default_ipc_fields;
 use super::ipc::{IpcField, IpcSchema};
 
+pub use super::ipc::write::default_ipc_fields;
+pub use crate::io::ipc::write::common::WriteOptions;
+
 /// Serializes [`Chunk`] to a vector of [`FlightData`] representing the serialized dictionaries
 /// and a [`FlightData`] representing the batch.
 /// # Errors
diff --git a/src/io/ipc/read/stream.rs b/src/io/ipc/read/stream.rs
index c49512faaa..3149e5e5b1 100644
--- a/src/io/ipc/read/stream.rs
+++ b/src/io/ipc/read/stream.rs
@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::io::Read;
 
 use arrow_format;
@@ -91,6 +92,7 @@ fn read_next<R: Read>(
     dictionaries: &mut Dictionaries,
     message_buffer: &mut Vec<u8>,
     data_buffer: &mut Vec<u8>,
+    projection: &Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
 ) -> Result<Option<StreamState>> {
     // determine metadata length
     let mut meta_length: [u8; 4] = [0; 4];
@@ -155,18 +157,26 @@ fn read_next<R: Read>(
 
             let mut reader = std::io::Cursor::new(data_buffer);
 
-            read_record_batch(
+            let chunk = read_record_batch(
                 batch,
                 &metadata.schema.fields,
                 &metadata.ipc_schema,
-                None,
+                projection.as_ref().map(|x| x.0.as_ref()),
                 dictionaries,
                 metadata.version,
                 &mut reader,
                 0,
                 file_size,
-            )
-            .map(|x| Some(StreamState::Some(x)))
+            );
+
+            if let Some((_, map, _)) = projection {
+                // re-order according to projection
+                chunk
+                    .map(|chunk| apply_projection(chunk, map))
+                    .map(|x| Some(StreamState::Some(x)))
+            } else {
+                chunk.map(|x| Some(StreamState::Some(x)))
+            }
         }
         arrow_format::ipc::MessageHeaderRef::DictionaryBatch(batch) => {
             let mut buf = vec![0; block_length];
@@ -185,7 +195,14 @@ fn read_next<R: Read>(
             )?;
 
             // read the next message until we encounter a RecordBatch message
-            read_next(reader, metadata, dictionaries, message_buffer, data_buffer)
+            read_next(
+                reader,
+                metadata,
+                dictionaries,
+                message_buffer,
+                data_buffer,
+                projection,
+            )
         }
         _ => Err(Error::from(OutOfSpecKind::UnexpectedMessageType)),
     }
@@ -204,6 +221,7 @@ pub struct StreamReader<R: Read> {
     finished: bool,
     data_buffer: Vec<u8>,
     message_buffer: Vec<u8>,
+    projection: Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
 }
 
 impl<R: Read> StreamReader<R> {
@@ -212,7 +230,16 @@ impl<R: Read> StreamReader<R> {
     /// The first message in the stream is the schema, the reader will fail if it does not
     /// encounter a schema.
     /// To check if the reader is done, use `is_finished(self)`
-    pub fn new(reader: R, metadata: StreamMetadata) -> Self {
+    pub fn new(reader: R, metadata: StreamMetadata, projection: Option<Vec<usize>>) -> Self {
+        let projection = projection.map(|projection| {
+            let (p, h, fields) = prepare_projection(&metadata.schema.fields, projection);
+            let schema = Schema {
+                fields,
+                metadata: metadata.schema.metadata.clone(),
+            };
+            (p, h, schema)
+        });
+
         Self {
             reader,
             metadata,
@@ -220,6 +247,7 @@ impl<R: Read> StreamReader<R> {
             finished: false,
             data_buffer: vec![],
             message_buffer: vec![],
+            projection,
         }
     }
 
@@ -228,6 +256,14 @@ impl<R: Read> StreamReader<R> {
         &self.metadata
     }
 
+    /// Return the schema of the file
+    pub fn schema(&self) -> &Schema {
+        self.projection
+            .as_ref()
+            .map(|x| &x.2)
+            .unwrap_or(&self.metadata.schema)
+    }
+
     /// Check if the stream is finished
     pub fn is_finished(&self) -> bool {
         self.finished
@@ -243,6 +279,7 @@ impl<R: Read> StreamReader<R> {
             &mut self.dictionaries,
             &mut self.message_buffer,
             &mut self.data_buffer,
+            &self.projection,
         )?;
         if batch.is_none() {
             self.finished = true;


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 58abc32ef9..40015100d0 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -88,7 +88,7 @@ jobs:
 
       - name: Run
         # --skip io: miri can't handle opening of files, so we skip those
-        run: cargo miri test --features full -- --skip io::parquet --skip io::ipc
+        run: cargo miri test --features full -- --skip io::parquet --skip io::ipc --skip io::flight
 
   miri-checks-io:
     name: MIRI on IO IPC
diff --git a/integration-testing/src/bin/arrow-stream-to-file.rs b/integration-testing/src/bin/arrow-stream-to-file.rs
index ab0855bf67..f5d9f81fd7 100644
--- a/integration-testing/src/bin/arrow-stream-to-file.rs
+++ b/integration-testing/src/bin/arrow-stream-to-file.rs
@@ -24,7 +24,7 @@ use arrow2::io::ipc::write;
 fn main() -> Result<()> {
     let mut reader = io::stdin();
     let metadata = read::read_stream_metadata(&mut reader)?;
-    let mut arrow_stream_reader = read::StreamReader::new(reader, metadata.clone());
+    let mut arrow_stream_reader = read::StreamReader::new(reader, metadata.clone(), None);
 
     let writer = io::stdout();
 
diff --git a/tests/it/io/flight/mod.rs b/tests/it/io/flight/mod.rs
new file mode 100644
index 0000000000..9718b74902
--- /dev/null
+++ b/tests/it/io/flight/mod.rs
@@ -0,0 +1,32 @@
+use arrow2::array::Array;
+use arrow2::chunk::Chunk;
+use arrow2::datatypes::Schema;
+use arrow2::error::Error;
+
+use arrow2::io::flight::*;
+use arrow2::io::ipc::write::{default_ipc_fields, WriteOptions};
+
+use super::ipc::read_gzip_json;
+
+fn round_trip(schema: Schema, chunk: Chunk<Box<dyn Array>>) -> Result<(), Error> {
+    let fields = default_ipc_fields(&schema.fields);
+    let serialized = serialize_schema(&schema, Some(&fields));
+    let (result, ipc_schema) = deserialize_schemas(&serialized.data_header)?;
+    assert_eq!(schema, result);
+
+    let (_, batch) = serialize_batch(&chunk, &fields, &WriteOptions { compression: None })?;
+
+    let result = deserialize_batch(&batch, &result.fields, &ipc_schema, &Default::default())?;
+    assert_eq!(result, chunk);
+    Ok(())
+}
+
+#[test]
+fn generated_nested_dictionary() -> Result<(), Error> {
+    let (schema, _, mut batches) =
+        read_gzip_json("1.0.0-littleendian", "generated_nested").unwrap();
+
+    round_trip(schema, batches.pop().unwrap())?;
+
+    Ok(())
+}
diff --git a/tests/it/io/ipc/common.rs b/tests/it/io/ipc/common.rs
index 45df69f46a..deaf44ec6a 100644
--- a/tests/it/io/ipc/common.rs
+++ b/tests/it/io/ipc/common.rs
@@ -46,7 +46,11 @@ pub fn read_gzip_json(version: &str, file_name: &str) -> Result<IpcRead> {
     Ok((schema, ipc_fields, batches))
 }
 
-pub fn read_arrow_stream(version: &str, file_name: &str) -> IpcRead {
+pub fn read_arrow_stream(
+    version: &str,
+    file_name: &str,
+    projection: Option<Vec<usize>>,
+) -> IpcRead {
     let testdata = crate::test_util::arrow_test_data();
     let mut file = File::open(format!(
         "{}/arrow-ipc-stream/integration/{}/{}.stream",
@@ -55,7 +59,7 @@ pub fn read_arrow_stream(version: &str, file_name: &str) -> IpcRead {
     .unwrap();
 
     let metadata = read_stream_metadata(&mut file).unwrap();
-    let reader = StreamReader::new(file, metadata);
+    let reader = StreamReader::new(file, metadata, projection);
 
     let schema = reader.metadata().schema.clone();
     let ipc_fields = reader.metadata().ipc_schema.fields.clone();
diff --git a/tests/it/io/ipc/read/stream.rs b/tests/it/io/ipc/read/stream.rs
index 18e7ecd075..ab8ae45a0e 100644
--- a/tests/it/io/ipc/read/stream.rs
+++ b/tests/it/io/ipc/read/stream.rs
@@ -1,3 +1,4 @@
+use arrow2::chunk::Chunk;
 use std::fs::File;
 
 use arrow2::error::Result;
@@ -13,7 +14,7 @@ fn test_file(version: &str, file_name: &str) -> Result<()> {
     ))?;
 
     let metadata = read_stream_metadata(&mut file)?;
-    let reader = StreamReader::new(file, metadata);
+    let reader = StreamReader::new(file, metadata, None);
 
     // read expected JSON output
     let (schema, ipc_fields, batches) = read_gzip_json(version, file_name)?;
@@ -100,3 +101,54 @@ fn read_generated_200_compression_lz4() -> Result<()> {
 fn read_generated_200_compression_zstd() -> Result<()> {
     test_file("2.0.0-compression", "generated_zstd")
 }
+
+fn test_projection(version: &str, file_name: &str, columns: Vec<usize>) -> Result<()> {
+    let testdata = crate::test_util::arrow_test_data();
+    let mut file = File::open(format!(
+        "{}/arrow-ipc-stream/integration/{}/{}.stream",
+        testdata, version, file_name
+    ))?;
+
+    let metadata = read_stream_metadata(&mut file)?;
+
+    let (_, _, chunks) = read_gzip_json(version, file_name)?;
+
+    let expected_fields = columns
+        .iter()
+        .copied()
+        .map(|x| metadata.schema.fields[x].clone())
+        .collect::<Vec<_>>();
+
+    let expected_chunks = chunks
+        .into_iter()
+        .map(|chunk| {
+            let columns = columns
+                .iter()
+                .copied()
+                .map(|x| chunk.arrays()[x].clone())
+                .collect::<Vec<_>>();
+            Chunk::new(columns)
+        })
+        .collect::<Vec<_>>();
+
+    let reader = StreamReader::new(&mut file, metadata, Some(columns.clone()));
+
+    assert_eq!(reader.schema().fields, expected_fields);
+
+    expected_chunks
+        .iter()
+        .zip(reader.map(|x| x.unwrap().unwrap()))
+        .for_each(|(lhs, rhs)| {
+            assert_eq!(lhs, &rhs);
+        });
+    Ok(())
+}
+
+#[test]
+fn read_projected() -> Result<()> {
+    test_projection("1.0.0-littleendian", "generated_primitive", vec![1])?;
+    test_projection("1.0.0-littleendian", "generated_dictionary", vec![2])?;
+    test_projection("1.0.0-littleendian", "generated_nested", vec![0])?;
+    test_projection("1.0.0-littleendian", "generated_primitive", vec![2, 1])?;
+    test_projection("1.0.0-littleendian", "generated_primitive", vec![0, 2, 1])
+}
diff --git a/tests/it/io/ipc/write/stream.rs b/tests/it/io/ipc/write/stream.rs
index 46f62aa11b..092e4402b7 100644
--- a/tests/it/io/ipc/write/stream.rs
+++ b/tests/it/io/ipc/write/stream.rs
@@ -30,13 +30,13 @@ fn write_(
 }
 
 fn test_file(version: &str, file_name: &str) {
-    let (schema, ipc_fields, batches) = read_arrow_stream(version, file_name);
+    let (schema, ipc_fields, batches) = read_arrow_stream(version, file_name, None);
 
     let result = write_(&schema, Some(ipc_fields), &batches);
 
     let mut reader = Cursor::new(result);
     let metadata = read_stream_metadata(&mut reader).unwrap();
-    let reader = StreamReader::new(reader, metadata);
+    let reader = StreamReader::new(reader, metadata, None);
 
     let schema = reader.metadata().schema.clone();
     let ipc_fields = reader.metadata().ipc_schema.fields.clone();
diff --git a/tests/it/io/ipc/write_file_async.rs b/tests/it/io/ipc/write_file_async.rs
index 7bd76b86be..25b8164f05 100644
--- a/tests/it/io/ipc/write_file_async.rs
+++ b/tests/it/io/ipc/write_file_async.rs
@@ -32,7 +32,7 @@ async fn write_(
 }
 
 async fn test_file(version: &str, file_name: &str) -> Result<()> {
-    let (schema, ipc_fields, batches) = read_arrow_stream(version, file_name);
+    let (schema, ipc_fields, batches) = read_arrow_stream(version, file_name, None);
 
     let result = write_(&schema, &ipc_fields, &batches).await?;
 
diff --git a/tests/it/io/ipc/write_stream_async.rs b/tests/it/io/ipc/write_stream_async.rs
index e033d5e943..12669bdd59 100644
--- a/tests/it/io/ipc/write_stream_async.rs
+++ b/tests/it/io/ipc/write_stream_async.rs
@@ -32,13 +32,13 @@ async fn write_(
 }
 
 async fn test_file(version: &str, file_name: &str) -> Result<()> {
-    let (schema, ipc_fields, batches) = read_arrow_stream(version, file_name);
+    let (schema, ipc_fields, batches) = read_arrow_stream(version, file_name, None);
 
     let result = write_(&schema, &ipc_fields, &batches).await?;
 
     let mut reader = Cursor::new(result);
     let metadata = read::read_stream_metadata(&mut reader)?;
-    let reader = read::StreamReader::new(reader, metadata);
+    let reader = read::StreamReader::new(reader, metadata, None);
 
     let schema = &reader.metadata().schema;
     let ipc_fields = reader.metadata().ipc_schema.fields.clone();
diff --git a/tests/it/io/mod.rs b/tests/it/io/mod.rs
index d192b806ee..3fc0c2123e 100644
--- a/tests/it/io/mod.rs
+++ b/tests/it/io/mod.rs
@@ -22,3 +22,6 @@ mod avro;
     feature = "io_csv_read_async"
 ))]
 mod csv;
+
+#[cfg(feature = "io_flight")]
+mod flight;