Repository: jorgecarleitao/arrow2
Pull request #1215 (2022-08-08 05:36:41)

Added support for parquet sidecar to `FileReader`
Initializing a `FileReader` currently:
* is IO bounded, by reading the metadata
* infers an Arrow schema (and fails if it can't infer one)
* clones the `FileMetadata` to have it available for users
* has an `Arc<Fn>` on it to filter row groups that most use-cases I saw so far is not useful

This PR refactors `FileReader::try_new` into `FileReader::new` by allowing its users more flexibility into how the file is read.

The main idea is that `FileReader` is now initialized as follows:

```rust
    // we can read its metadata:
    let metadata = read::read_metadata(&mut reader)?;

    // and infer a [`Schema`] from the `metadata`.
    let schema = read::infer_schema(&metadata)?;

    // we can filter the columns we need (here we select all)
    // (projection pushdown)
    let schema = schema.filter(|_index, _field| true);

    // we can read the statistics of all parquet's row groups (here for the first field)
    let statistics = read::statistics::deserialize(&schema.fields[0], &metadata.row_groups)?;

    // say we found that we only need to read the first two row groups, "0" and "1"
    // (row-group filter pushdown)
    let row_groups = metadata
        .row_groups
        .into_iter()
        .enumerate()
        .filter(|(index, _)| *index == 0 || *index == 1)
        .map(|(_, row_group)| row_group)
        .collect();

    // we can then read the row groups into chunks
    let chunks = read::FileReader::new(reader, row_groups, schema, Some(1024 * 8 * 8), None);
```

this gives the user the flexibility to perform the necessary preparation to the metadata (`Schema` and `row_groups`) to read the file. `let row_groups = metadata.row_groups` if no predictive pushdown is being used.

This also allows row groups defined somewhere else (e.g. a parquet sidecar) to be used.

codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#1215](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (a2b9198) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/497d4310ede92cc8acd223fcedce0ef2763e09eb?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (497d431) will **decrease** coverage by `0.19%`.
> The diff coverage is `96.96%`.

```diff
@@            Coverage Diff             @@
##             main    #1215      +/-   ##
==========================================
- Coverage   83.49%   83.29%   -0.20%     
==========================================
  Files         356      358       +2     
  Lines       37081    37682     +601     
==========================================
+ Hits        30961    31389     +428     
- Misses       6120     6293     +173     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/io/parquet/read/file.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvcmVhZC9maWxlLnJz) | `90.62% <93.75%> (+6.25%)` | :arrow_up: |
| [src/datatypes/schema.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2RhdGF0eXBlcy9zY2hlbWEucnM=) | `80.00% <100.00%> (+26.15%)` | :arrow_up: |
| [src/io/flight/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2ZsaWdodC9tb2QucnM=) | `40.51% <0.00%> (-26.49%)` | :arrow_down: |
| [src/io/ipc/read/array/boolean.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL2FycmF5L2Jvb2xlYW4ucnM=) | `90.74% <0.00%> (-7.41%)` | :arrow_down: |
| [src/io/parquet/write/sink.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvc2luay5ycw==) | `71.42% <0.00%> (-0.75%)` | :arrow_down: |
| [src/io/ipc/read/file.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL2ZpbGUucnM=) | `97.76% <0.00%> (ø)` | |
| [src/array/struct\_/mutable.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L3N0cnVjdF8vbXV0YWJsZS5ycw==) | `38.88% <0.00%> (ø)` | |
| [src/io/ipc/read/common.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL2NvbW1vbi5ycw==) | `95.29% <0.00%> (+0.39%)` | :arrow_up: |
| [src/io/json/read/infer\_schema.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2pzb24vcmVhZC9pbmZlcl9zY2hlbWEucnM=) | `96.08% <0.00%> (+0.60%)` | :arrow_up: |
| [src/bitmap/immutable.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2JpdG1hcC9pbW11dGFibGUucnM=) | `86.41% <0.00%> (+0.61%)` | :arrow_up: |
| ... and [8 more](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1215/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)


Patch:
diff --git a/benches/read_parquet.rs b/benches/read_parquet.rs
index 3d1dc85858..94c089931a 100644
--- a/benches/read_parquet.rs
+++ b/benches/read_parquet.rs
@@ -32,10 +32,16 @@ fn to_buffer(
     buffer
 }
 
-fn read_batch(buffer: &[u8], size: usize, column: usize) -> Result<()> {
-    let file = Cursor::new(buffer);
+fn read_chunk(buffer: &[u8], size: usize, column: usize) -> Result<()> {
+    let mut reader = Cursor::new(buffer);
 
-    let reader = read::FileReader::try_new(file, Some(&[column]), None, None, None)?;
+    let metadata = read::read_metadata(&mut reader)?;
+
+    let schema = read::infer_schema(&metadata)?;
+
+    let schema = schema.filter(|index, _| index == column);
+
+    let reader = read::FileReader::new(reader, metadata.row_groups, schema, None, None);
 
     for maybe_chunk in reader {
         let columns = maybe_chunk?;
@@ -49,43 +55,43 @@ fn add_benchmark(c: &mut Criterion) {
         let size = 2usize.pow(i);
         let buffer = to_buffer(size, true, false, false, false);
         let a = format!("read i64 2^{}", i);
-        c.bench_function(&a, |b| b.iter(|| read_batch(&buffer, size, 0).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| read_chunk(&buffer, size, 0).unwrap()));
 
         let a = format!("read utf8 2^{}", i);
-        c.bench_function(&a, |b| b.iter(|| read_batch(&buffer, size, 2).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| read_chunk(&buffer, size, 2).unwrap()));
 
         let a = format!("read utf8 large 2^{}", i);
-        c.bench_function(&a, |b| b.iter(|| read_batch(&buffer, size, 6).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| read_chunk(&buffer, size, 6).unwrap()));
 
         let a = format!("read utf8 emoji 2^{}", i);
-        c.bench_function(&a, |b| b.iter(|| read_batch(&buffer, size, 12).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| read_chunk(&buffer, size, 12).unwrap()));
 
         let a = format!("read bool 2^{}", i);
-        c.bench_function(&a, |b| b.iter(|| read_batch(&buffer, size, 3).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| read_chunk(&buffer, size, 3).unwrap()));
 
         let buffer = to_buffer(size, true, true, false, false);
         let a = format!("read utf8 dict 2^{}", i);
-        c.bench_function(&a, |b| b.iter(|| read_batch(&buffer, size, 2).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| read_chunk(&buffer, size, 2).unwrap()));
 
         let buffer = to_buffer(size, true, false, false, true);
         let a = format!("read i64 snappy 2^{}", i);
-        c.bench_function(&a, |b| b.iter(|| read_batch(&buffer, size, 0).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| read_chunk(&buffer, size, 0).unwrap()));
 
         let buffer = to_buffer(size, true, false, true, false);
         let a = format!("read utf8 multi 2^{}", i);
-        c.bench_function(&a, |b| b.iter(|| read_batch(&buffer, size, 2).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| read_chunk(&buffer, size, 2).unwrap()));
 
         let buffer = to_buffer(size, true, false, true, true);
         let a = format!("read utf8 multi snappy 2^{}", i);
-        c.bench_function(&a, |b| b.iter(|| read_batch(&buffer, size, 2).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| read_chunk(&buffer, size, 2).unwrap()));
 
         let buffer = to_buffer(size, true, false, true, true);
         let a = format!("read i64 multi snappy 2^{}", i);
-        c.bench_function(&a, |b| b.iter(|| read_batch(&buffer, size, 0).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| read_chunk(&buffer, size, 0).unwrap()));
 
         let buffer = to_buffer(size, false, false, false, false);
         let a = format!("read required utf8 2^{}", i);
-        c.bench_function(&a, |b| b.iter(|| read_batch(&buffer, size, 2).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| read_chunk(&buffer, size, 2).unwrap()));
     });
 }
 
diff --git a/examples/parquet_read.rs b/examples/parquet_read.rs
index 92fa83c521..e0e5a220e8 100644
--- a/examples/parquet_read.rs
+++ b/examples/parquet_read.rs
@@ -1,30 +1,44 @@
 use std::fs::File;
 use std::time::SystemTime;
 
-use arrow2::error::Result;
+use arrow2::error::Error;
 use arrow2::io::parquet::read;
 
-fn main() -> Result<()> {
+fn main() -> Result<(), Error> {
+    // say we have a file
     use std::env;
     let args: Vec<String> = env::args().collect();
-
     let file_path = &args[1];
+    let mut reader = File::open(file_path)?;
 
-    let reader = File::open(file_path)?;
-    let reader = read::FileReader::try_new(reader, None, Some(1024 * 8 * 8), None, None)?;
+    // we can read its metadata:
+    let metadata = read::read_metadata(&mut reader)?;
 
-    println!("{:#?}", reader.schema());
+    // and infer a [`Schema`] from the `metadata`.
+    let schema = read::infer_schema(&metadata)?;
 
-    // say we want to evaluate if the we can skip some row groups based on a field's value
-    let field = &reader.schema().fields[0];
+    // we can filter the columns we need (here we select all)
+    let schema = schema.filter(|_index, _field| true);
 
-    // we can deserialize the parquet statistics from this field
-    let statistics = read::statistics::deserialize(field, &reader.metadata().row_groups)?;
+    // we can read the statistics of all parquet's row groups (here for the first field)
+    let statistics = read::statistics::deserialize(&schema.fields[0], &metadata.row_groups)?;
 
     println!("{:#?}", statistics);
 
+    // say we found that we only need to read the first two row groups, "0" and "1"
+    let row_groups = metadata
+        .row_groups
+        .into_iter()
+        .enumerate()
+        .filter(|(index, _)| *index == 0 || *index == 1)
+        .map(|(_, row_group)| row_group)
+        .collect();
+
+    // we can then read the row groups into chunks
+    let chunks = read::FileReader::new(reader, row_groups, schema, Some(1024 * 8 * 8), None);
+
     let start = SystemTime::now();
-    for maybe_chunk in reader {
+    for maybe_chunk in chunks {
         let chunk = maybe_chunk?;
         assert!(!chunk.is_empty());
     }
diff --git a/examples/parquet_read_async.rs b/examples/parquet_read_async.rs
index 2428b36886..3f1abc4faf 100644
--- a/examples/parquet_read_async.rs
+++ b/examples/parquet_read_async.rs
@@ -43,7 +43,7 @@ async fn main() -> Result<()> {
         // the runtime.
         // Furthermore, this operation is trivially paralellizable e.g. via rayon, as each iterator
         // can be advanced in parallel (parallel decompression and deserialization).
-        let chunks = RowGroupDeserializer::new(column_chunks, row_group.num_rows() as usize, None);
+        let chunks = RowGroupDeserializer::new(column_chunks, row_group.num_rows(), None);
         for maybe_chunk in chunks {
             let chunk = maybe_chunk?;
             println!("{}", chunk.len());
diff --git a/examples/s3/src/main.rs b/examples/s3/src/main.rs
index 5aa24431dd..d3f668d442 100644
--- a/examples/s3/src/main.rs
+++ b/examples/s3/src/main.rs
@@ -71,7 +71,7 @@ async fn main() -> Result<()> {
 
     // this is CPU-bounded and should be sent to a separate thread-pool.
     // We do it here for simplicity
-    let chunks = read::RowGroupDeserializer::new(column_chunks, group.num_rows() as usize, None);
+    let chunks = read::RowGroupDeserializer::new(column_chunks, group.num_rows(), None);
     let chunks = chunks.collect::<Result<Vec<_>>>()?;
 
     // this is a single chunk because chunk_size is `None`
diff --git a/src/datatypes/schema.rs b/src/datatypes/schema.rs
index baa0447636..671c943862 100644
--- a/src/datatypes/schema.rs
+++ b/src/datatypes/schema.rs
@@ -26,6 +26,28 @@ impl Schema {
             metadata,
         }
     }
+
+    /// Returns a new [`Schema`] with a subset of all fields whose `predicate`
+    /// evaluates to true.
+    pub fn filter<F: Fn(usize, &Field) -> bool>(self, predicate: F) -> Self {
+        let fields = self
+            .fields
+            .into_iter()
+            .enumerate()
+            .filter_map(|(index, f)| {
+                if (predicate)(index, &f) {
+                    Some(f)
+                } else {
+                    None
+                }
+            })
+            .collect();
+
+        Schema {
+            fields,
+            metadata: self.metadata,
+        }
+    }
 }
 
 impl From<Vec<Field>> for Schema {
diff --git a/src/io/parquet/read/file.rs b/src/io/parquet/read/file.rs
index 638089668a..ff8944d38c 100644
--- a/src/io/parquet/read/file.rs
+++ b/src/io/parquet/read/file.rs
@@ -1,18 +1,12 @@
 use std::io::{Read, Seek};
-use std::sync::Arc;
 
 use crate::array::Array;
 use crate::chunk::Chunk;
 use crate::datatypes::Schema;
+use crate::error::Result;
 use crate::io::parquet::read::read_columns_many;
-use crate::{
-    datatypes::Field,
-    error::{Error, Result},
-};
 
-use super::{infer_schema, read_metadata, FileMetaData, RowGroupDeserializer, RowGroupMetaData};
-
-type GroupFilter = Arc<dyn Fn(usize, &RowGroupMetaData) -> bool + Send + Sync>;
+use super::{RowGroupDeserializer, RowGroupMetaData};
 
 /// An iterator of [`Chunk`]s coming from row groups of a parquet file.
 ///
@@ -20,96 +14,29 @@ type GroupFilter = Arc<dyn Fn(usize, &RowGroupMetaData) -> bool + Send + Sync>;
 /// mapped to an [`Iterator<Item=Chunk>`] and each iterator is iterated upon until either the limit
 /// or the last iterator ends.
 /// # Implementation
-/// This iterator mixes IO-bounded and CPU-bounded operations.
+/// This iterator is single threaded on both IO-bounded and CPU-bounded tasks, and mixes them.
 pub struct FileReader<R: Read + Seek> {
     row_groups: RowGroupReader<R>,
-    metadata: FileMetaData,
     remaining_rows: usize,
     current_row_group: Option<RowGroupDeserializer>,
 }
 
 impl<R: Read + Seek> FileReader<R> {
-    /// Creates a new [`FileReader`] by reading the metadata from `reader` and constructing
-    /// Arrow's schema from it.
-    ///
-    /// # Error
-    /// This function errors iff:
-    /// * reading the metadata from the reader fails
-    /// * it is not possible to derive an arrow schema from the parquet file
-    /// * the projection contains columns that do not exist
-    pub fn try_new(
-        mut reader: R,
-        projection: Option<&[usize]>,
+    /// Returns a new [`FileReader`].
+    pub fn new(
+        reader: R,
+        row_groups: Vec<RowGroupMetaData>,
+        schema: Schema,
         chunk_size: Option<usize>,
         limit: Option<usize>,
-        groups_filter: Option<GroupFilter>,
-    ) -> Result<Self> {
-        let metadata = read_metadata(&mut reader)?;
-
-        let schema = infer_schema(&metadata)?;
-
-        let schema_metadata = schema.metadata;
-        let fields: Vec<Field> = if let Some(projection) = &projection {
-            schema
-                .fields
-                .into_iter()
-                .enumerate()
-                .filter_map(|(index, f)| {
-                    if projection.iter().any(|&i| i == index) {
-                        Some(f)
-                    } else {
-                        None
-                    }
-                })
-                .collect()
-        } else {
-            schema.fields.into_iter().collect()
-        };
-
-        if let Some(projection) = &projection {
-            if fields.len() != projection.len() {
-                return Err(Error::InvalidArgumentError(
-                    "While reading parquet, some columns in the projection do not exist in the file"
-                        .to_string(),
-                ));
-            }
-        }
-
-        let schema = Schema {
-            fields,
-            metadata: schema_metadata,
-        };
-
-        let row_groups = RowGroupReader::new(
-            reader,
-            schema,
-            groups_filter,
-            metadata.row_groups.clone(),
-            chunk_size,
-            limit,
-        );
+    ) -> Self {
+        let row_groups = RowGroupReader::new(reader, schema, row_groups, chunk_size, limit);
 
-        Ok(Self {
+        Self {
             row_groups,
-            metadata,
             remaining_rows: limit.unwrap_or(usize::MAX),
             current_row_group: None,
-        })
-    }
-
-    /// Returns the derived arrow [`Schema`] of the file
-    pub fn schema(&self) -> &Schema {
-        &self.row_groups.schema
-    }
-
-    /// Returns parquet's [`FileMetaData`].
-    pub fn metadata(&self) -> &FileMetaData {
-        &self.metadata
-    }
-
-    /// Sets the groups filter
-    pub fn set_groups_filter(&mut self, groups_filter: GroupFilter) {
-        self.row_groups.set_groups_filter(groups_filter);
+        }
     }
 
     fn next_row_group(&mut self) -> Result<Option<RowGroupDeserializer>> {
@@ -178,7 +105,6 @@ impl<R: Read + Seek> Iterator for FileReader<R> {
 pub struct RowGroupReader<R: Read + Seek> {
     reader: R,
     schema: Schema,
-    groups_filter: Option<GroupFilter>,
     row_groups: std::iter::Enumerate<std::vec::IntoIter<RowGroupMetaData>>,
     chunk_size: Option<usize>,
     remaining_rows: usize,
@@ -189,7 +115,6 @@ impl<R: Read + Seek> RowGroupReader<R> {
     pub fn new(
         reader: R,
         schema: Schema,
-        groups_filter: Option<GroupFilter>,
         row_groups: Vec<RowGroupMetaData>,
         chunk_size: Option<usize>,
         limit: Option<usize>,
@@ -197,18 +122,12 @@ impl<R: Read + Seek> RowGroupReader<R> {
         Self {
             reader,
             schema,
-            groups_filter,
             row_groups: row_groups.into_iter().enumerate(),
             chunk_size,
             remaining_rows: limit.unwrap_or(usize::MAX),
         }
     }
 
-    /// Sets the groups filter
-    pub fn set_groups_filter(&mut self, groups_filter: GroupFilter) {
-        self.groups_filter = Some(groups_filter);
-    }
-
     #[inline]
     fn _next(&mut self) -> Result<Option<RowGroupDeserializer>> {
         if self.schema.fields.is_empty() {
@@ -219,14 +138,7 @@ impl<R: Read + Seek> RowGroupReader<R> {
             return Ok(None);
         }
 
-        let row_group = if let Some(groups_filter) = self.groups_filter.as_ref() {
-            self.row_groups
-                .by_ref()
-                .find(|(index, row_group)| (groups_filter)(*index, row_group))
-        } else {
-            self.row_groups.next()
-        };
-        let row_group = if let Some((_, row_group)) = row_group {
+        let row_group = if let Some((_, row_group)) = self.row_groups.next() {
             row_group
         } else {
             return Ok(None);
@@ -242,12 +154,10 @@ impl<R: Read + Seek> RowGroupReader<R> {
 
         let result = RowGroupDeserializer::new(
             column_chunks,
-            row_group.num_rows() as usize,
+            row_group.num_rows(),
             Some(self.remaining_rows),
         );
-        self.remaining_rows = self
-            .remaining_rows
-            .saturating_sub(row_group.num_rows() as usize);
+        self.remaining_rows = self.remaining_rows.saturating_sub(row_group.num_rows());
         Ok(Some(result))
     }
 }
diff --git a/src/io/parquet/read/row_group.rs b/src/io/parquet/read/row_group.rs
index 4dde21e467..59fba886a6 100644
--- a/src/io/parquet/read/row_group.rs
+++ b/src/io/parquet/read/row_group.rs
@@ -260,8 +260,6 @@ pub async fn read_columns_many_async<
     field_columns
         .into_iter()
         .zip(fields.into_iter())
-        .map(|(columns, field)| {
-            to_deserializer(columns, field, row_group.num_rows() as usize, chunk_size)
-        })
+        .map(|(columns, field)| to_deserializer(columns, field, row_group.num_rows(), chunk_size))
         .collect()
 }


Test patch:
diff --git a/tests/it/io/parquet/mod.rs b/tests/it/io/parquet/mod.rs
index 8b8a3c2038..55baceaa9a 100644
--- a/tests/it/io/parquet/mod.rs
+++ b/tests/it/io/parquet/mod.rs
@@ -7,8 +7,8 @@ use arrow2::{
     chunk::Chunk,
     datatypes::*,
     error::Result,
+    io::parquet::read as p_read,
     io::parquet::read::statistics::*,
-    io::parquet::read::*,
     io::parquet::write::*,
     types::{days_ms, NativeType},
 };
@@ -23,29 +23,24 @@ mod write_async;
 type ArrayStats = (Box<dyn Array>, Statistics);
 
 pub fn read_column<R: Read + Seek>(mut reader: R, column: &str) -> Result<ArrayStats> {
-    let metadata = read_metadata(&mut reader)?;
-    let schema = infer_schema(&metadata)?;
+    let metadata = p_read::read_metadata(&mut reader)?;
+    let schema = p_read::infer_schema(&metadata)?;
 
     // verify that we can read indexes
-    let _indexes = read_columns_indexes(
+    let _indexes = p_read::read_columns_indexes(
         &mut reader,
         metadata.row_groups[0].columns(),
         &schema.fields,
     )?;
 
-    let column = schema
-        .fields
-        .iter()
-        .enumerate()
-        .find_map(|(i, f)| if f.name == column { Some(i) } else { None })
-        .unwrap();
+    let schema = schema.filter(|_, f| f.name == column);
 
-    let mut reader = FileReader::try_new(reader, Some(&[column]), None, None, None)?;
-
-    let field = &schema.fields[column];
+    let field = &schema.fields[0];
 
     let statistics = deserialize(field, &metadata.row_groups)?;
 
+    let mut reader = p_read::FileReader::new(reader, metadata.row_groups, schema, None, None);
+
     Ok((
         reader.next().unwrap()?.into_arrays().pop().unwrap(),
         statistics,
@@ -1150,13 +1145,22 @@ fn integration_write(schema: &Schema, chunks: &[Chunk<Box<dyn Array>>]) -> Resul
 type IntegrationRead = (Schema, Vec<Chunk<Box<dyn Array>>>);
 
 fn integration_read(data: &[u8], limit: Option<usize>) -> Result<IntegrationRead> {
-    let reader = FileReader::try_new(Cursor::new(data), None, None, limit, None)?;
-    let schema = reader.schema().clone();
+    let mut reader = Cursor::new(data);
+    let metadata = p_read::read_metadata(&mut reader)?;
+    let schema = p_read::infer_schema(&metadata)?;
 
     for field in &schema.fields {
-        let mut _statistics = deserialize(field, &reader.metadata().row_groups)?;
+        let mut _statistics = deserialize(field, &metadata.row_groups)?;
     }
 
+    let reader = p_read::FileReader::new(
+        Cursor::new(data),
+        metadata.row_groups,
+        schema.clone(),
+        None,
+        limit,
+    );
+
     let batches = reader.collect::<Result<Vec<_>>>()?;
 
     Ok((schema, batches))
@@ -1527,23 +1531,26 @@ fn filter_chunk() -> Result<()> {
 
     let r = integration_write(&schema, &[chunk1.clone(), chunk2.clone()])?;
 
-    let reader = FileReader::try_new(
-        Cursor::new(r),
-        None,
-        None,
-        None,
-        // select chunk 1
-        Some(std::sync::Arc::new(|i, _| i == 0)),
-    )?;
-    let new_schema = reader.schema().clone();
+    let mut reader = Cursor::new(r);
 
-    for field in &schema.fields {
-        let mut _statistics = deserialize(field, &reader.metadata().row_groups)?;
-    }
+    let metadata = p_read::read_metadata(&mut reader)?;
+
+    let new_schema = p_read::infer_schema(&metadata)?;
+    assert_eq!(new_schema, schema);
+
+    // select chunk 1
+    let row_groups = metadata
+        .row_groups
+        .into_iter()
+        .enumerate()
+        .filter(|(index, _)| *index == 0)
+        .map(|(_, row_group)| row_group)
+        .collect();
+
+    let reader = p_read::FileReader::new(reader, row_groups, schema, None, None);
 
     let new_chunks = reader.collect::<Result<Vec<_>>>()?;
 
-    assert_eq!(new_schema, schema);
     assert_eq!(new_chunks, vec![chunk1]);
     Ok(())
 }
diff --git a/tests/it/io/parquet/read.rs b/tests/it/io/parquet/read.rs
index 45c5b714d6..efb3943da9 100644
--- a/tests/it/io/parquet/read.rs
+++ b/tests/it/io/parquet/read.rs
@@ -496,9 +496,11 @@ fn v1_map_nullable() -> Result<()> {
 #[test]
 fn all_types() -> Result<()> {
     let path = "testing/parquet-testing/data/alltypes_plain.parquet";
-    let reader = std::fs::File::open(path)?;
+    let mut reader = std::fs::File::open(path)?;
 
-    let reader = FileReader::try_new(reader, None, None, None, None)?;
+    let metadata = read_metadata(&mut reader)?;
+    let schema = infer_schema(&metadata)?;
+    let reader = FileReader::new(reader, metadata.row_groups, schema, None, None);
 
     let batches = reader.collect::<Result<Vec<_>>>()?;
     assert_eq!(batches.len(), 1);
@@ -535,10 +537,12 @@ fn all_types() -> Result<()> {
 fn all_types_chunked() -> Result<()> {
     // this has one batch with 8 elements
     let path = "testing/parquet-testing/data/alltypes_plain.parquet";
-    let reader = std::fs::File::open(path)?;
+    let mut reader = std::fs::File::open(path)?;
 
+    let metadata = read_metadata(&mut reader)?;
+    let schema = infer_schema(&metadata)?;
     // chunk it in 5 (so, (5,3))
-    let reader = FileReader::try_new(reader, None, Some(5), None, None)?;
+    let reader = FileReader::new(reader, metadata.row_groups, schema, Some(5), None);
 
     let batches = reader.collect::<Result<Vec<_>>>()?;
     assert_eq!(batches.len(), 2);
@@ -584,7 +588,7 @@ fn all_types_chunked() -> Result<()> {
 
 #[cfg(feature = "io_parquet_compression")]
 #[test]
-fn invalid_utf8() {
+fn invalid_utf8() -> Result<()> {
     let invalid_data = &[
         0x50, 0x41, 0x52, 0x31, 0x15, 0x00, 0x15, 0x24, 0x15, 0x28, 0x2c, 0x15, 0x02, 0x15, 0x00,
         0x15, 0x06, 0x15, 0x08, 0x00, 0x00, 0x12, 0x44, 0x02, 0x00, 0x00, 0x00, 0x03, 0xff, 0x08,
@@ -597,8 +601,11 @@ fn invalid_utf8() {
         0x42, 0x00, 0x51, 0x00, 0x00, 0x00, 0x50, 0x41, 0x52, 0x31,
     ];
 
-    let reader = Cursor::new(invalid_data);
-    let reader = FileReader::try_new(reader, None, Some(5), None, None).unwrap();
+    let mut reader = Cursor::new(invalid_data);
+
+    let metadata = read_metadata(&mut reader)?;
+    let schema = infer_schema(&metadata)?;
+    let reader = FileReader::new(reader, metadata.row_groups, schema, Some(5), None);
 
     let error = reader.collect::<Result<Vec<_>>>().unwrap_err();
     assert!(
@@ -606,4 +613,5 @@ fn invalid_utf8() {
         "unexpected error: {}",
         error
     );
+    Ok(())
 }
diff --git a/tests/it/io/parquet/read_indexes.rs b/tests/it/io/parquet/read_indexes.rs
index 68f09bae80..328a826eaf 100644
--- a/tests/it/io/parquet/read_indexes.rs
+++ b/tests/it/io/parquet/read_indexes.rs
@@ -125,7 +125,7 @@ fn read_with_indexes(
         vec![&c1.descriptor().descriptor.primitive_type],
         schema.fields[1].clone(),
         None,
-        row_group.num_rows() as usize,
+        row_group.num_rows(),
     )?;
 
     let arrays = arrays.collect::<Result<Vec<_>>>()?;
diff --git a/tests/it/io/parquet/write_async.rs b/tests/it/io/parquet/write_async.rs
index 5644caf67f..000c39ca64 100644
--- a/tests/it/io/parquet/write_async.rs
+++ b/tests/it/io/parquet/write_async.rs
@@ -63,7 +63,7 @@ async fn test_parquet_async_roundtrip() {
         let column_chunks = read_columns_many_async(factory, group, schema.fields.clone(), None)
             .await
             .unwrap();
-        let chunks = RowGroupDeserializer::new(column_chunks, group.num_rows() as usize, None);
+        let chunks = RowGroupDeserializer::new(column_chunks, group.num_rows(), None);
         let mut chunks = chunks.collect::<Result<Vec<_>>>().unwrap();
         out.append(&mut chunks);
     }