Repository: jorgecarleitao/arrow2
Pull request #1139 (2022-07-04 14:17:16)

Fixed panic in deserializing nested statistics
When deserializing nested statistics from parquet, we need to take into account that the null count of a `List[Struct]` can't be `UInt64` and should instead be something like `List[{"a": null_a, "b": null_b}]`. This PR fixes this.

I think we can still improve this a bit (e.g. why not `{"a": null_a, "b": null_b}` instead?) so that users can use nested statistics to create predicates for filter pushdown of nested types, but at least this no longer panics.



Patch:
diff --git a/LICENSE b/LICENSE
index f93c66a762..4d9d7075a7 100644
--- a/LICENSE
+++ b/LICENSE
@@ -1,13 +1,191 @@
-Copyright [2021] [Jorge C Leitao]
 
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
+                                 Apache License
+                           Version 2.0, January 2004
+                        https://www.apache.org/licenses/
 
-    http://www.apache.org/licenses/LICENSE-2.0
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
 
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   Copyright 2020-2022 Jorge C. Leitão
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       https://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
diff --git a/examples/extension.rs b/examples/extension.rs
index 158b9bbd43..c0aea84316 100644
--- a/examples/extension.rs
+++ b/examples/extension.rs
@@ -51,6 +51,6 @@ fn write_ipc<W: Write + Seek>(writer: W, array: impl Array + 'static) -> Result<
 fn read_ipc(buf: &[u8]) -> Result<Chunk<Box<dyn Array>>> {
     let mut cursor = Cursor::new(buf);
     let metadata = read::read_file_metadata(&mut cursor)?;
-    let mut reader = read::FileReader::new(cursor, metadata, None);
+    let mut reader = read::FileReader::new(cursor, metadata, None, None);
     reader.next().unwrap()
 }
diff --git a/examples/ipc_file_read.rs b/examples/ipc_file_read.rs
index 18ceed0ba8..c171a5d18d 100644
--- a/examples/ipc_file_read.rs
+++ b/examples/ipc_file_read.rs
@@ -9,7 +9,7 @@ use arrow2::io::print;
 
 /// Simplest way: read all record batches from the file. This can be used e.g. for random access.
 #[allow(clippy::type_complexity)]
-fn read_batches(path: &str) -> Result<(Schema, Vec<Chunk<Box<dyn Array>>>)> {
+fn read_chunks(path: &str) -> Result<(Schema, Vec<Chunk<Box<dyn Array>>>)> {
     let mut file = File::open(path)?;
 
     // read the files' metadata. At this point, we can distribute the read whatever we like.
@@ -18,10 +18,10 @@ fn read_batches(path: &str) -> Result<(Schema, Vec<Chunk<Box<dyn Array>>>)> {
     let schema = metadata.schema.clone();
 
     // Simplest way: use the reader, an iterator over batches.
-    let reader = read::FileReader::new(file, metadata, None);
+    let reader = read::FileReader::new(file, metadata, None, None);
 
-    let columns = reader.collect::<Result<Vec<_>>>()?;
-    Ok((schema, columns))
+    let chunks = reader.collect::<Result<Vec<_>>>()?;
+    Ok((schema, chunks))
 }
 
 /// Random access way: read a single record batch from the file. This can be used e.g. for random access.
@@ -36,6 +36,7 @@ fn read_batch(path: &str) -> Result<(Schema, Chunk<Box<dyn Array>>)> {
     // advanced way: read the dictionary
     let dictionaries = read::read_file_dictionaries(&mut file, &metadata, &mut Default::default())?;
 
+    // and the chunk
     let chunk_index = 0;
 
     let chunk = read::read_batch(
@@ -43,6 +44,7 @@ fn read_batch(path: &str) -> Result<(Schema, Chunk<Box<dyn Array>>)> {
         &dictionaries,
         &metadata,
         None,
+        None,
         chunk_index,
         &mut Default::default(),
         &mut Default::default(),
@@ -57,12 +59,12 @@ fn main() -> Result<()> {
 
     let file_path = &args[1];
 
-    let (schema, batches) = read_batches(file_path)?;
+    let (schema, chunks) = read_chunks(file_path)?;
     let names = schema.fields.iter().map(|f| &f.name).collect::<Vec<_>>();
-    println!("{}", print::write(&batches, &names));
+    println!("{}", print::write(&chunks, &names));
 
-    let (schema, batch) = read_batch(file_path)?;
+    let (schema, chunk) = read_batch(file_path)?;
     let names = schema.fields.iter().map(|f| &f.name).collect::<Vec<_>>();
-    println!("{}", print::write(&[batch], &names));
+    println!("{}", print::write(&[chunk], &names));
     Ok(())
 }
diff --git a/examples/ipc_file_write.rs b/examples/ipc_file_write.rs
index 74c7ffeedf..51f570b228 100644
--- a/examples/ipc_file_write.rs
+++ b/examples/ipc_file_write.rs
@@ -6,15 +6,15 @@ use arrow2::datatypes::{DataType, Field, Schema};
 use arrow2::error::Result;
 use arrow2::io::ipc::write;
 
-fn write_batches(path: &str, schema: Schema, columns: &[Chunk<Box<dyn Array>>]) -> Result<()> {
+fn write_batches(path: &str, schema: Schema, chunks: &[Chunk<Box<dyn Array>>]) -> Result<()> {
     let file = File::create(path)?;
 
     let options = write::WriteOptions { compression: None };
     let mut writer = write::FileWriter::new(file, schema, None, options);
 
     writer.start()?;
-    for columns in columns {
-        writer.write(columns, None)?
+    for chunk in chunks {
+        writer.write(chunk, None)?
     }
     writer.finish()
 }
@@ -34,9 +34,9 @@ fn main() -> Result<()> {
     let a = Int32Array::from_slice(&[1, 2, 3, 4, 5]);
     let b = Utf8Array::<i32>::from_slice(&["a", "b", "c", "d", "e"]);
 
-    let batch = Chunk::try_new(vec![a.boxed(), b.boxed()])?;
+    let chunk = Chunk::try_new(vec![a.boxed(), b.boxed()])?;
 
     // write it
-    write_batches(file_path, schema, &[batch])?;
+    write_batches(file_path, schema, &[chunk])?;
     Ok(())
 }
diff --git a/examples/parquet_write_parallel/src/main.rs b/examples/parquet_write_parallel/src/main.rs
index 3d6a0fe0d4..28cf299310 100644
--- a/examples/parquet_write_parallel/src/main.rs
+++ b/examples/parquet_write_parallel/src/main.rs
@@ -58,7 +58,7 @@ fn parallel_write(path: &str, schema: Schema, chunks: &[Chunk]) -> Result<()> {
 
     // declare encodings
     let encodings = (&schema.fields)
-        .par_iter()
+        .iter()
         .map(|f| transverse(&f.data_type, encoding_map))
         .collect::<Vec<_>>();
 
diff --git a/license.py b/license.py
deleted file mode 100644
index f60ca4d722..0000000000
--- a/license.py
+++ /dev/null
@@ -1,101 +0,0 @@
-"""
-Adds licenses haders to all Rust source files (.rs). This script
-is idempotent and errors if any file has more than one license.
-"""
-import os
-
-# Because some code was copied from the arrow project,
-# we should keep the original license on those portions.
-asf_license = """// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements.  See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership.  The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License.  You may obtain a copy of the License at
-//
-//   http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied.  See the License for the
-// specific language governing permissions and limitations
-// under the License."""
-
-with open('LICENSE', "r") as f:
-    own_license = f.read()
-own_license = '// ' + own_license.replace('\n', '\n// ')[:-3]
-
-
-own_compute = [
-    "src/compute/arity.rs",
-    "src/compute/hash.rs",
-    "src/compute/boolean_kleene.rs",
-    "src/compute/merge_sort",
-]
-
-def should_be_own_license(path: str) -> bool:
-    return path.startswith("src/array") or \
-        path.startswith("src/buffer") or \
-        path.startswith("src/bitmap") or \
-        path.startswith("src/error") or \
-        path.startswith("src/lib.rs") or \
-        path.startswith("src/types") or \
-        path.startswith("src/docs") or \
-        path.startswith("src/bits") or \
-        path == "src/temporal_conversions.rs" or \
-        path.startswith("src/io/csv") or \
-        path.startswith("src/io/parquet") or \
-        any(path.startswith(x) for x in own_compute)
-
-
-def should_be_asf_license(path: str) -> bool:
-    return path.startswith("src/alloc") or \
-        (path.startswith("src/compute") and not any(path.startswith(x) for x in own_compute)) or \
-        path.startswith("src/io") and not path.startswith("src/io/csv") or \
-        path == "src/record_batch.rs" or \
-        path.startswith("src/datatypes/") or path.startswith("src/ffi/") or \
-        path.startswith("src/util/")
-
-
-for root, subdirs, files in os.walk('src/'):
-    for filename in files:
-        if filename.endswith(".rs"):
-            file_path = os.path.join(root, filename)
-            has_own_license = False
-            has_asf_license = False
-            with open(file_path, "r") as f:
-                contents = f.read()
-
-            if contents[:len(own_license)] == own_license:
-                has_own_license = True
-            if contents[:len(asf_license)] == asf_license:
-                has_asf_license = True
-
-            should_be_own = should_be_own_license(file_path)
-            should_be_asf = should_be_asf_license(file_path)
-            if not should_be_own and not should_be_asf:
-                print(f"needs some: {file_path}")
-                exit(1)
-            if should_be_own and should_be_asf:
-                print(f"needs both: {file_path}")
-                exit(1)
-
-            if should_be_own:
-                if has_own_license:
-                    print(f"Removing own license from {file_path}")
-                    contents = contents[len(own_license):]
-                elif has_asf_license:
-                    print(f"Removing asf license from {file_path}")
-                    contents = contents[len(asf_license):]
-                if has_asf_license or has_own_license:
-                    with open(file_path, "w") as f:
-                        f.write(contents)
-            elif should_be_asf:
-                if has_own_license:
-                    contents = contents[len(own_license):]
-                if not has_asf_license or has_own_license:
-                    print(f"Adding ASF license to {file_path}")
-                    with open(file_path, "w") as f:
-                        f.write(asf_license + '\n\n' + contents)
diff --git a/src/io/flight/mod.rs b/src/io/flight/mod.rs
index 9ed0835bb8..59ba0a83d8 100644
--- a/src/io/flight/mod.rs
+++ b/src/io/flight/mod.rs
@@ -137,6 +137,7 @@ pub fn deserialize_batch(
             fields,
             ipc_schema,
             None,
+            None,
             dictionaries,
             message.version()?,
             &mut reader,
diff --git a/src/io/ipc/mod.rs b/src/io/ipc/mod.rs
index 68124c8bfa..3d8a8ccb99 100644
--- a/src/io/ipc/mod.rs
+++ b/src/io/ipc/mod.rs
@@ -58,10 +58,10 @@
 //! // Fetch some of the data and get the reader back
 //! let mut reader = File::open(&path)?;
 //! let metadata = read_file_metadata(&mut reader)?;
-//! let mut filereader = FileReader::new(reader, metadata, None);
-//! let row1 = filereader.next().unwrap();  // [[-1, 1], [1, -1]]
-//! let row2 = filereader.next().unwrap();  // [[-1, 1], [1, -1]]
-//! let mut reader = filereader.into_inner();
+//! let mut reader = FileReader::new(reader, metadata, None, None);
+//! let row1 = reader.next().unwrap();  // [[-1, 1], [1, -1]]
+//! let row2 = reader.next().unwrap();  // [[-1, 1], [1, -1]]
+//! let mut reader = reader.into_inner();
 //! // Do more stuff with the reader, like seeking ahead.
 //! # Ok::<(), Error>(())
 //! ```
diff --git a/src/io/ipc/read/array/binary.rs b/src/io/ipc/read/array/binary.rs
index f32ca3e871..3ee6c4ba28 100644
--- a/src/io/ipc/read/array/binary.rs
+++ b/src/io/ipc/read/array/binary.rs
@@ -18,6 +18,7 @@ pub fn read_binary<O: Offset, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    limit: Option<usize>,
     scratch: &mut Vec<u8>,
 ) -> Result<BinaryArray<O>> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
@@ -34,6 +35,7 @@ pub fn read_binary<O: Offset, R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        limit,
         scratch,
     )?;
 
@@ -41,6 +43,7 @@ pub fn read_binary<O: Offset, R: Read + Seek>(
         .length()
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+    let length = limit.map(|limit| limit.min(length)).unwrap_or(length);
 
     let offsets: Buffer<O> = read_buffer(
         buffers,
@@ -54,7 +57,7 @@ pub fn read_binary<O: Offset, R: Read + Seek>(
     // Older versions of the IPC format sometimes do not report an offset
     .or_else(|_| Result::Ok(Buffer::<O>::from(vec![O::default()])))?;
 
-    let last_offset = offsets.as_slice()[offsets.len() - 1].to_usize();
+    let last_offset = offsets.last().unwrap().to_usize();
     let values = read_buffer(
         buffers,
         last_offset,
diff --git a/src/io/ipc/read/array/boolean.rs b/src/io/ipc/read/array/boolean.rs
index 4440062a29..f53c0afd99 100644
--- a/src/io/ipc/read/array/boolean.rs
+++ b/src/io/ipc/read/array/boolean.rs
@@ -17,6 +17,7 @@ pub fn read_boolean<R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    limit: Option<usize>,
     scratch: &mut Vec<u8>,
 ) -> Result<BooleanArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
@@ -26,11 +27,6 @@ pub fn read_boolean<R: Read + Seek>(
         ))
     })?;
 
-    let length: usize = field_node
-        .length()
-        .try_into()
-        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
-
     let validity = read_validity(
         buffers,
         field_node,
@@ -38,9 +34,16 @@ pub fn read_boolean<R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        limit,
         scratch,
     )?;
 
+    let length: usize = field_node
+        .length()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+    let length = limit.map(|limit| limit.min(length)).unwrap_or(length);
+
     let values = read_bitmap(
         buffers,
         length,
diff --git a/src/io/ipc/read/array/dictionary.rs b/src/io/ipc/read/array/dictionary.rs
index 63a0cb0b80..69c617852f 100644
--- a/src/io/ipc/read/array/dictionary.rs
+++ b/src/io/ipc/read/array/dictionary.rs
@@ -18,6 +18,7 @@ pub fn read_dictionary<T: DictionaryKey, R: Read + Seek>(
     dictionaries: &Dictionaries,
     block_offset: u64,
     compression: Option<Compression>,
+    limit: Option<usize>,
     is_little_endian: bool,
     scratch: &mut Vec<u8>,
 ) -> Result<DictionaryArray<T>>
@@ -48,6 +49,7 @@ where
         block_offset,
         is_little_endian,
         compression,
+        limit,
         scratch,
     )?;
 
diff --git a/src/io/ipc/read/array/fixed_size_binary.rs b/src/io/ipc/read/array/fixed_size_binary.rs
index ffceb00bc9..72e0bd8f5b 100644
--- a/src/io/ipc/read/array/fixed_size_binary.rs
+++ b/src/io/ipc/read/array/fixed_size_binary.rs
@@ -17,6 +17,7 @@ pub fn read_fixed_size_binary<R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    limit: Option<usize>,
     scratch: &mut Vec<u8>,
 ) -> Result<FixedSizeBinaryArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
@@ -33,6 +34,7 @@ pub fn read_fixed_size_binary<R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        limit,
         scratch,
     )?;
 
@@ -40,6 +42,7 @@ pub fn read_fixed_size_binary<R: Read + Seek>(
         .length()
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+    let length = limit.map(|limit| limit.min(length)).unwrap_or(length);
 
     let length = length.saturating_mul(FixedSizeBinaryArray::maybe_get_size(&data_type)?);
     let values = read_buffer(
diff --git a/src/io/ipc/read/array/fixed_size_list.rs b/src/io/ipc/read/array/fixed_size_list.rs
index 3ea4ef42b3..94c1c7662c 100644
--- a/src/io/ipc/read/array/fixed_size_list.rs
+++ b/src/io/ipc/read/array/fixed_size_list.rs
@@ -21,6 +21,7 @@ pub fn read_fixed_size_list<R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    limit: Option<usize>,
     version: Version,
     scratch: &mut Vec<u8>,
 ) -> Result<FixedSizeListArray> {
@@ -38,10 +39,13 @@ pub fn read_fixed_size_list<R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        limit,
         scratch,
     )?;
 
-    let (field, _) = FixedSizeListArray::get_child_and_size(&data_type);
+    let (field, size) = FixedSizeListArray::get_child_and_size(&data_type);
+
+    let limit = limit.map(|x| x.saturating_mul(size));
 
     let values = read(
         field_nodes,
@@ -53,6 +57,7 @@ pub fn read_fixed_size_list<R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        limit,
         version,
         scratch,
     )?;
diff --git a/src/io/ipc/read/array/list.rs b/src/io/ipc/read/array/list.rs
index e12dc5e02e..7870b85a7b 100644
--- a/src/io/ipc/read/array/list.rs
+++ b/src/io/ipc/read/array/list.rs
@@ -23,6 +23,7 @@ pub fn read_list<O: Offset, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    limit: Option<usize>,
     version: Version,
     scratch: &mut Vec<u8>,
 ) -> Result<ListArray<O>>
@@ -43,6 +44,7 @@ where
         block_offset,
         is_little_endian,
         compression,
+        limit,
         scratch,
     )?;
 
@@ -50,6 +52,7 @@ where
         .length()
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+    let length = limit.map(|limit| limit.min(length)).unwrap_or(length);
 
     let offsets = read_buffer::<O, _>(
         buffers,
@@ -63,6 +66,8 @@ where
     // Older versions of the IPC format sometimes do not report an offset
     .or_else(|_| Result::Ok(Buffer::<O>::from(vec![O::default()])))?;
 
+    let last_offset = offsets.last().unwrap().to_usize();
+
     let field = ListArray::<O>::get_child_field(&data_type);
 
     let values = read(
@@ -75,6 +80,7 @@ where
         block_offset,
         is_little_endian,
         compression,
+        Some(last_offset),
         version,
         scratch,
     )?;
diff --git a/src/io/ipc/read/array/map.rs b/src/io/ipc/read/array/map.rs
index 8a2a5e4a12..b0803ca5f2 100644
--- a/src/io/ipc/read/array/map.rs
+++ b/src/io/ipc/read/array/map.rs
@@ -22,6 +22,7 @@ pub fn read_map<R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    limit: Option<usize>,
     version: Version,
     scratch: &mut Vec<u8>,
 ) -> Result<MapArray> {
@@ -39,6 +40,7 @@ pub fn read_map<R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        limit,
         scratch,
     )?;
 
@@ -46,6 +48,7 @@ pub fn read_map<R: Read + Seek>(
         .length()
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+    let length = limit.map(|limit| limit.min(length)).unwrap_or(length);
 
     let offsets = read_buffer::<i32, _>(
         buffers,
@@ -61,6 +64,8 @@ pub fn read_map<R: Read + Seek>(
 
     let field = MapArray::get_field(&data_type);
 
+    let last_offset: usize = offsets.last().copied().unwrap() as usize;
+
     let field = read(
         field_nodes,
         field,
@@ -71,6 +76,7 @@ pub fn read_map<R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        Some(last_offset),
         version,
         scratch,
     )?;
diff --git a/src/io/ipc/read/array/primitive.rs b/src/io/ipc/read/array/primitive.rs
index 434aab0bd9..bba155f78e 100644
--- a/src/io/ipc/read/array/primitive.rs
+++ b/src/io/ipc/read/array/primitive.rs
@@ -17,6 +17,7 @@ pub fn read_primitive<T: NativeType, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    limit: Option<usize>,
     scratch: &mut Vec<u8>,
 ) -> Result<PrimitiveArray<T>>
 where
@@ -36,6 +37,7 @@ where
         block_offset,
         is_little_endian,
         compression,
+        limit,
         scratch,
     )?;
 
@@ -43,6 +45,7 @@ where
         .length()
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+    let length = limit.map(|limit| limit.min(length)).unwrap_or(length);
 
     let values = read_buffer(
         buffers,
diff --git a/src/io/ipc/read/array/struct_.rs b/src/io/ipc/read/array/struct_.rs
index 3bbd777a01..77f5d74bed 100644
--- a/src/io/ipc/read/array/struct_.rs
+++ b/src/io/ipc/read/array/struct_.rs
@@ -21,6 +21,7 @@ pub fn read_struct<R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    limit: Option<usize>,
     version: Version,
     scratch: &mut Vec<u8>,
 ) -> Result<StructArray> {
@@ -38,6 +39,7 @@ pub fn read_struct<R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        limit,
         scratch,
     )?;
 
@@ -57,6 +59,7 @@ pub fn read_struct<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                limit,
                 version,
                 scratch,
             )
diff --git a/src/io/ipc/read/array/union.rs b/src/io/ipc/read/array/union.rs
index 678d65ded9..627d912b44 100644
--- a/src/io/ipc/read/array/union.rs
+++ b/src/io/ipc/read/array/union.rs
@@ -22,6 +22,7 @@ pub fn read_union<R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    limit: Option<usize>,
     version: Version,
     scratch: &mut Vec<u8>,
 ) -> Result<UnionArray> {
@@ -42,6 +43,7 @@ pub fn read_union<R: Read + Seek>(
         .length()
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+    let length = limit.map(|limit| limit.min(length)).unwrap_or(length);
 
     let types = read_buffer(
         buffers,
@@ -87,6 +89,7 @@ pub fn read_union<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                None,
                 version,
                 scratch,
             )
diff --git a/src/io/ipc/read/array/utf8.rs b/src/io/ipc/read/array/utf8.rs
index 342da0c4de..f57e51ecc5 100644
--- a/src/io/ipc/read/array/utf8.rs
+++ b/src/io/ipc/read/array/utf8.rs
@@ -18,6 +18,7 @@ pub fn read_utf8<O: Offset, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    limit: Option<usize>,
     scratch: &mut Vec<u8>,
 ) -> Result<Utf8Array<O>> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
@@ -34,6 +35,7 @@ pub fn read_utf8<O: Offset, R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        limit,
         scratch,
     )?;
 
@@ -42,6 +44,8 @@ pub fn read_utf8<O: Offset, R: Read + Seek>(
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
 
+    let length = limit.map(|limit| limit.min(length)).unwrap_or(length);
+
     let offsets: Buffer<O> = read_buffer(
         buffers,
         1 + length,
@@ -54,7 +58,7 @@ pub fn read_utf8<O: Offset, R: Read + Seek>(
     // Older versions of the IPC format sometimes do not report an offset
     .or_else(|_| Result::Ok(Buffer::<O>::from(vec![O::default()])))?;
 
-    let last_offset = offsets.as_slice()[offsets.len() - 1].to_usize();
+    let last_offset = offsets.last().unwrap().to_usize();
     let values = read_buffer(
         buffers,
         last_offset,
diff --git a/src/io/ipc/read/common.rs b/src/io/ipc/read/common.rs
index 26252dd183..fe445f2079 100644
--- a/src/io/ipc/read/common.rs
+++ b/src/io/ipc/read/common.rs
@@ -71,7 +71,7 @@ impl<'a, A, I: Iterator<Item = A>> Iterator for ProjectionIter<'a, A, I> {
     }
 }
 
-/// Creates a record batch from binary data using the `ipc::RecordBatch` indexes and the `Schema`
+/// Returns a [`Chunk`] from a reader.
 /// # Panic
 /// Panics iff the projection is not in increasing order (e.g. `[1, 0]` nor `[0, 1, 1]` are valid)
 #[allow(clippy::too_many_arguments)]
@@ -80,6 +80,7 @@ pub fn read_record_batch<R: Read + Seek>(
     fields: &[Field],
     ipc_schema: &IpcSchema,
     projection: Option<&[usize]>,
+    limit: Option<usize>,
     dictionaries: &Dictionaries,
     version: arrow_format::ipc::MetadataVersion,
     reader: &mut R,
@@ -136,6 +137,7 @@ pub fn read_record_batch<R: Read + Seek>(
                     batch.compression().map_err(|err| {
                         Error::from(OutOfSpecKind::InvalidFlatbufferCompression(err))
                     })?,
+                    limit,
                     version,
                     scratch,
                 )?)),
@@ -163,6 +165,7 @@ pub fn read_record_batch<R: Read + Seek>(
                     batch.compression().map_err(|err| {
                         Error::from(OutOfSpecKind::InvalidFlatbufferCompression(err))
                     })?,
+                    limit,
                     version,
                     scratch,
                 )
@@ -270,6 +273,7 @@ pub fn read_dictionary<R: Read + Seek>(
                 &fields,
                 &ipc_schema,
                 None,
+                None, // we must read the whole dictionary
                 dictionaries,
                 arrow_format::ipc::MetadataVersion::V5,
                 reader,
diff --git a/src/io/ipc/read/deserialize.rs b/src/io/ipc/read/deserialize.rs
index 780f33e0d5..fec5676def 100644
--- a/src/io/ipc/read/deserialize.rs
+++ b/src/io/ipc/read/deserialize.rs
@@ -23,6 +23,7 @@ pub fn read<R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<BodyCompressionRef>,
+    limit: Option<usize>,
     version: MetadataVersion,
     scratch: &mut Vec<u8>,
 ) -> Result<Box<dyn Array>> {
@@ -39,6 +40,7 @@ pub fn read<R: Read + Seek>(
             block_offset,
             is_little_endian,
             compression,
+            limit,
             scratch,
         )
         .map(|x| x.boxed()),
@@ -51,6 +53,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                limit,
                 scratch,
             )
             .map(|x| x.boxed())
@@ -64,6 +67,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                limit,
                 scratch,
             )?;
             Ok(Box::new(array))
@@ -77,6 +81,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                limit,
                 scratch,
             )?;
             Ok(Box::new(array))
@@ -90,6 +95,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                limit,
                 scratch,
             )?;
             Ok(Box::new(array))
@@ -103,6 +109,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                limit,
                 scratch,
             )?;
             Ok(Box::new(array))
@@ -116,6 +123,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                limit,
                 scratch,
             )?;
             Ok(Box::new(array))
@@ -130,6 +138,7 @@ pub fn read<R: Read + Seek>(
             block_offset,
             is_little_endian,
             compression,
+            limit,
             version,
             scratch,
         )
@@ -144,6 +153,7 @@ pub fn read<R: Read + Seek>(
             block_offset,
             is_little_endian,
             compression,
+            limit,
             version,
             scratch,
         )
@@ -158,6 +168,7 @@ pub fn read<R: Read + Seek>(
             block_offset,
             is_little_endian,
             compression,
+            limit,
             version,
             scratch,
         )
@@ -172,6 +183,7 @@ pub fn read<R: Read + Seek>(
             block_offset,
             is_little_endian,
             compression,
+            limit,
             version,
             scratch,
         )
@@ -186,8 +198,9 @@ pub fn read<R: Read + Seek>(
                     dictionaries,
                     block_offset,
                     compression,
+                    limit,
                     is_little_endian,
-                    scratch
+                    scratch,
                 )
                 .map(|x| x.boxed())
             })
@@ -202,6 +215,7 @@ pub fn read<R: Read + Seek>(
             block_offset,
             is_little_endian,
             compression,
+            limit,
             version,
             scratch,
         )
@@ -216,6 +230,7 @@ pub fn read<R: Read + Seek>(
             block_offset,
             is_little_endian,
             compression,
+            limit,
             version,
             scratch,
         )
diff --git a/src/io/ipc/read/file_async.rs b/src/io/ipc/read/file_async.rs
index 100990a650..0d80b80ac6 100644
--- a/src/io/ipc/read/file_async.rs
+++ b/src/io/ipc/read/file_async.rs
@@ -31,7 +31,12 @@ impl<'a> FileStream<'a> {
     ///
     /// # Examples
     /// See [`FileSink`](crate::io::ipc::write::file_async::FileSink).
-    pub fn new<R>(reader: R, metadata: FileMetadata, projection: Option<Vec<usize>>) -> Self
+    pub fn new<R>(
+        reader: R,
+        metadata: FileMetadata,
+        projection: Option<Vec<usize>>,
+        limit: Option<usize>,
+    ) -> Self
     where
         R: AsyncRead + AsyncSeek + Unpin + Send + 'a,
     {
@@ -46,7 +51,7 @@ impl<'a> FileStream<'a> {
             (None, None)
         };
 
-        let stream = Self::stream(reader, None, metadata.clone(), projection);
+        let stream = Self::stream(reader, None, metadata.clone(), projection, limit);
         Self {
             stream,
             metadata,
@@ -69,6 +74,7 @@ impl<'a> FileStream<'a> {
         mut dictionaries: Option<Dictionaries>,
         metadata: FileMetadata,
         projection: Option<(Vec<usize>, HashMap<usize, usize>)>,
+        limit: Option<usize>,
     ) -> BoxStream<'a, Result<Chunk<Box<dyn Array>>>>
     where
         R: AsyncRead + AsyncSeek + Unpin + Send + 'a,
@@ -80,17 +86,20 @@ impl<'a> FileStream<'a> {
             let mut meta_buffer = Default::default();
             let mut block_buffer = Default::default();
             let mut scratch = Default::default();
+            let mut remaining = limit.unwrap_or(usize::MAX);
             for block in 0..metadata.blocks.len() {
                 let chunk = read_batch(
                     &mut reader,
                     dictionaries.as_mut().unwrap(),
                     &metadata,
                     projection.as_ref().map(|x| x.0.as_ref()),
+                    Some(remaining),
                     block,
                     &mut meta_buffer,
                     &mut block_buffer,
                     &mut scratch
                 ).await?;
+                remaining -= chunk.len();
 
                 let chunk = if let Some((_, map)) = &projection {
                     // re-order according to projection
@@ -159,6 +168,7 @@ async fn read_batch<R>(
     dictionaries: &mut Dictionaries,
     metadata: &FileMetadata,
     projection: Option<&[usize]>,
+    limit: Option<usize>,
     block: usize,
     meta_buffer: &mut Vec<u8>,
     block_buffer: &mut Vec<u8>,
@@ -217,6 +227,7 @@ where
         &metadata.schema.fields,
         &metadata.ipc_schema,
         projection,
+        limit,
         dictionaries,
         message
             .version()
diff --git a/src/io/ipc/read/read_basic.rs b/src/io/ipc/read/read_basic.rs
index e58b01b58e..0a93a63a21 100644
--- a/src/io/ipc/read/read_basic.rs
+++ b/src/io/ipc/read/read_basic.rs
@@ -254,6 +254,7 @@ pub fn read_bitmap<R: Read + Seek>(
     Bitmap::try_new(buffer, length)
 }
 
+#[allow(clippy::too_many_arguments)]
 pub fn read_validity<R: Read + Seek>(
     buffers: &mut VecDeque<IpcBuffer>,
     field_node: Node,
@@ -261,12 +262,14 @@ pub fn read_validity<R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    limit: Option<usize>,
     scratch: &mut Vec<u8>,
 ) -> Result<Option<Bitmap>> {
     let length: usize = field_node
         .length()
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+    let length = limit.map(|limit| limit.min(length)).unwrap_or(length);
 
     Ok(if field_node.null_count() > 0 {
         Some(read_bitmap(
diff --git a/src/io/ipc/read/reader.rs b/src/io/ipc/read/reader.rs
index 801f5041ae..b18ca4cead 100644
--- a/src/io/ipc/read/reader.rs
+++ b/src/io/ipc/read/reader.rs
@@ -36,18 +36,6 @@ pub struct FileMetadata {
     pub(crate) size: u64,
 }
 
-/// Arrow File reader
-pub struct FileReader<R: Read + Seek> {
-    reader: R,
-    metadata: FileMetadata,
-    // the dictionaries are going to be read
-    dictionaries: Option<Dictionaries>,
-    current_block: usize,
-    projection: Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
-    data_scratch: Vec<u8>,
-    message_scratch: Vec<u8>,
-}
-
 fn read_dictionary_message<R: Read + Seek>(
     reader: &mut R,
     offset: u64,
@@ -262,11 +250,13 @@ pub(super) fn get_serialized_batch<'a>(
 /// certain parts of the file.
 /// # Panics
 /// This function panics iff `index >= metadata.blocks.len()`
+#[allow(clippy::too_many_arguments)]
 pub fn read_batch<R: Read + Seek>(
     reader: &mut R,
     dictionaries: &Dictionaries,
     metadata: &FileMetadata,
     projection: Option<&[usize]>,
+    limit: Option<usize>,
     index: usize,
     message_scratch: &mut Vec<u8>,
     data_scratch: &mut Vec<u8>,
@@ -317,6 +307,7 @@ pub fn read_batch<R: Read + Seek>(
         &metadata.schema.fields,
         &metadata.ipc_schema,
         projection,
+        limit,
         dictionaries,
         message
             .version()
@@ -328,11 +319,29 @@ pub fn read_batch<R: Read + Seek>(
     )
 }
 
+/// An iterator of [`Chunk`]s from an Arrow IPC file.
+pub struct FileReader<R: Read + Seek> {
+    reader: R,
+    metadata: FileMetadata,
+    // the dictionaries are going to be read
+    dictionaries: Option<Dictionaries>,
+    current_block: usize,
+    projection: Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
+    remaining: usize,
+    data_scratch: Vec<u8>,
+    message_scratch: Vec<u8>,
+}
+
 impl<R: Read + Seek> FileReader<R> {
     /// Creates a new [`FileReader`]. Use `projection` to only take certain columns.
     /// # Panic
     /// Panics iff the projection is not in increasing order (e.g. `[1, 0]` nor `[0, 1, 1]` are valid)
-    pub fn new(reader: R, metadata: FileMetadata, projection: Option<Vec<usize>>) -> Self {
+    pub fn new(
+        reader: R,
+        metadata: FileMetadata,
+        projection: Option<Vec<usize>>,
+        limit: Option<usize>,
+    ) -> Self {
         let projection = projection.map(|projection| {
             let (p, h, fields) = prepare_projection(&metadata.schema.fields, projection);
             let schema = Schema {
@@ -346,6 +355,7 @@ impl<R: Read + Seek> FileReader<R> {
             metadata,
             dictionaries: Default::default(),
             projection,
+            remaining: limit.unwrap_or(usize::MAX),
             current_block: 0,
             data_scratch: Default::default(),
             message_scratch: Default::default(),
@@ -404,10 +414,12 @@ impl<R: Read + Seek> Iterator for FileReader<R> {
             self.dictionaries.as_ref().unwrap(),
             &self.metadata,
             self.projection.as_ref().map(|x| x.0.as_ref()),
+            Some(self.remaining),
             block,
             &mut self.message_scratch,
             &mut self.data_scratch,
         );
+        self.remaining -= chunk.as_ref().map(|x| x.len()).unwrap_or_default();
 
         let chunk = if let Some((_, map, _)) = &self.projection {
             // re-order according to projection
diff --git a/src/io/ipc/read/stream.rs b/src/io/ipc/read/stream.rs
index e7d3d590aa..b3f3a986d6 100644
--- a/src/io/ipc/read/stream.rs
+++ b/src/io/ipc/read/stream.rs
@@ -173,6 +173,7 @@ fn read_next<R: Read>(
                 &metadata.schema.fields,
                 &metadata.ipc_schema,
                 projection.as_ref().map(|x| x.0.as_ref()),
+                None,
                 dictionaries,
                 metadata.version,
                 &mut reader,
diff --git a/src/io/ipc/read/stream_async.rs b/src/io/ipc/read/stream_async.rs
index b637dc0a2d..4c8556f0ea 100644
--- a/src/io/ipc/read/stream_async.rs
+++ b/src/io/ipc/read/stream_async.rs
@@ -143,6 +143,7 @@ async fn maybe_next<R: AsyncRead + Unpin + Send>(
                 &state.metadata.schema.fields,
                 &state.metadata.ipc_schema,
                 None,
+                None,
                 &state.dictionaries,
                 state.metadata.version,
                 &mut std::io::Cursor::new(&state.data_buffer),
diff --git a/src/io/ipc/write/file_async.rs b/src/io/ipc/write/file_async.rs
index dfe0c0b7fc..3ee7e7ee89 100644
--- a/src/io/ipc/write/file_async.rs
+++ b/src/io/ipc/write/file_async.rs
@@ -54,7 +54,7 @@ type WriteOutput<W> = (usize, Option<Block>, Vec<Block>, Option<W>);
 /// // Read chunks from file
 /// buffer.set_position(0);
 /// let metadata = read_file_metadata_async(&mut buffer).await?;
-/// let mut stream = FileStream::new(buffer, metadata, None);
+/// let mut stream = FileStream::new(buffer, metadata, None, None);
 /// let chunks = stream.try_collect::<Vec<_>>().await?;
 /// # arrow2::error::Result::Ok(())
 /// # }).unwrap();
diff --git a/src/io/parquet/read/statistics/mod.rs b/src/io/parquet/read/statistics/mod.rs
index ff9449f0be..d27123fc51 100644
--- a/src/io/parquet/read/statistics/mod.rs
+++ b/src/io/parquet/read/statistics/mod.rs
@@ -34,8 +34,12 @@ use super::get_field_columns;
 /// Enum of a count statistics
 #[derive(Debug, PartialEq)]
 pub enum Count {
-    /// simple arrays (every type not a Struct) have a count of UInt64
+    /// simple arrays have a count of UInt64
     Single(UInt64Array),
+    /// list arrays have a count as a list of UInt64
+    List(ListArray<i32>),
+    /// list arrays have a count as a list of UInt64
+    LargeList(ListArray<i64>),
     /// struct arrays have a count as a struct of UInt64
     Struct(StructArray),
     /// map arrays have a count as a map of UInt64
@@ -88,6 +92,24 @@ impl From<MutableStatistics> for Statistics {
                 .unwrap()
                 .clone();
             Count::Map(a)
+        } else if let PhysicalType::List = s.null_count.data_type().to_physical_type() {
+            let a = s
+                .null_count
+                .as_box()
+                .as_any()
+                .downcast_ref::<ListArray<i32>>()
+                .unwrap()
+                .clone();
+            Count::List(a)
+        } else if let PhysicalType::LargeList = s.null_count.data_type().to_physical_type() {
+            let a = s
+                .null_count
+                .as_box()
+                .as_any()
+                .downcast_ref::<ListArray<i64>>()
+                .unwrap()
+                .clone();
+            Count::LargeList(a)
         } else {
             let a = s
                 .null_count
@@ -98,35 +120,54 @@ impl From<MutableStatistics> for Statistics {
                 .clone();
             Count::Single(a)
         };
-        let distinct_count =
-            if let PhysicalType::Struct = s.distinct_count.data_type().to_physical_type() {
-                let a = s
-                    .distinct_count
-                    .as_box()
-                    .as_any()
-                    .downcast_ref::<StructArray>()
-                    .unwrap()
-                    .clone();
-                Count::Struct(a)
-            } else if let PhysicalType::Map = s.null_count.data_type().to_physical_type() {
-                let a = s
-                    .null_count
-                    .as_box()
-                    .as_any()
-                    .downcast_ref::<MapArray>()
-                    .unwrap()
-                    .clone();
-                Count::Map(a)
-            } else {
-                let a = s
-                    .distinct_count
-                    .as_box()
-                    .as_any()
-                    .downcast_ref::<UInt64Array>()
-                    .unwrap()
-                    .clone();
-                Count::Single(a)
-            };
+        let distinct_count = if let PhysicalType::Struct =
+            s.distinct_count.data_type().to_physical_type()
+        {
+            let a = s
+                .distinct_count
+                .as_box()
+                .as_any()
+                .downcast_ref::<StructArray>()
+                .unwrap()
+                .clone();
+            Count::Struct(a)
+        } else if let PhysicalType::Map = s.distinct_count.data_type().to_physical_type() {
+            let a = s
+                .distinct_count
+                .as_box()
+                .as_any()
+                .downcast_ref::<MapArray>()
+                .unwrap()
+                .clone();
+            Count::Map(a)
+        } else if let PhysicalType::List = s.distinct_count.data_type().to_physical_type() {
+            let a = s
+                .distinct_count
+                .as_box()
+                .as_any()
+                .downcast_ref::<ListArray<i32>>()
+                .unwrap()
+                .clone();
+            Count::List(a)
+        } else if let PhysicalType::LargeList = s.distinct_count.data_type().to_physical_type() {
+            let a = s
+                .distinct_count
+                .as_box()
+                .as_any()
+                .downcast_ref::<ListArray<i64>>()
+                .unwrap()
+                .clone();
+            Count::LargeList(a)
+        } else {
+            let a = s
+                .distinct_count
+                .as_box()
+                .as_any()
+                .downcast_ref::<UInt64Array>()
+                .unwrap()
+                .clone();
+            Count::Single(a)
+        };
         Self {
             null_count,
             distinct_count,
@@ -198,6 +239,18 @@ fn create_dt(data_type: &DataType) -> DataType {
             Box::new(Field::new(&f.name, create_dt(&f.data_type), f.is_nullable)),
             *ordered,
         )
+    } else if let DataType::List(f) = data_type.to_logical_type() {
+        DataType::List(Box::new(Field::new(
+            &f.name,
+            create_dt(&f.data_type),
+            f.is_nullable,
+        )))
+    } else if let DataType::LargeList(f) = data_type.to_logical_type() {
+        DataType::LargeList(Box::new(Field::new(
+            &f.name,
+            create_dt(&f.data_type),
+            f.is_nullable,
+        )))
     } else {
         DataType::UInt64
     }
@@ -301,12 +354,20 @@ fn push(
                 .as_mut_any()
                 .downcast_mut::<list::DynMutableListArray>()
                 .unwrap();
+            let distinct_count = distinct_count
+                .as_mut_any()
+                .downcast_mut::<list::DynMutableListArray>()
+                .unwrap();
+            let null_count = null_count
+                .as_mut_any()
+                .downcast_mut::<list::DynMutableListArray>()
+                .unwrap();
             return push(
                 stats,
                 min.inner.as_mut(),
                 max.inner.as_mut(),
-                distinct_count,
-                null_count,
+                distinct_count.inner.as_mut(),
+                null_count.inner.as_mut(),
             );
         }
         Dictionary(_, _, _) => {


Test patch:
diff --git a/integration-testing/src/bin/arrow-file-to-stream.rs b/integration-testing/src/bin/arrow-file-to-stream.rs
index ae8ed094fd..c63606e28c 100644
--- a/integration-testing/src/bin/arrow-file-to-stream.rs
+++ b/integration-testing/src/bin/arrow-file-to-stream.rs
@@ -32,7 +32,7 @@ fn main() -> Result<()> {
     let filename = args.file_name;
     let mut f = File::open(filename)?;
     let metadata = read::read_file_metadata(&mut f)?;
-    let mut reader = read::FileReader::new(f, metadata.clone(), None);
+    let mut reader = read::FileReader::new(f, metadata.clone(), None, None);
 
     let options = write::WriteOptions { compression: None };
     let mut writer = write::StreamWriter::new(std::io::stdout(), options);
diff --git a/integration-testing/src/bin/arrow-json-integration-test.rs b/integration-testing/src/bin/arrow-json-integration-test.rs
index 7152fb40a7..31f58a4ff2 100644
--- a/integration-testing/src/bin/arrow-json-integration-test.rs
+++ b/integration-testing/src/bin/arrow-json-integration-test.rs
@@ -80,7 +80,7 @@ fn arrow_to_json(arrow_name: &str, json_name: &str, verbose: bool) -> Result<()>
 
     let mut arrow_file = File::open(arrow_name)?;
     let metadata = read::read_file_metadata(&mut arrow_file)?;
-    let reader = read::FileReader::new(arrow_file, metadata.clone(), None);
+    let reader = read::FileReader::new(arrow_file, metadata.clone(), None, None);
 
     let names = metadata
         .schema
@@ -118,7 +118,7 @@ fn validate(arrow_name: &str, json_name: &str, verbose: bool) -> Result<()> {
     // open Arrow file
     let mut arrow_file = File::open(arrow_name)?;
     let metadata = read::read_file_metadata(&mut arrow_file)?;
-    let reader = read::FileReader::new(arrow_file, metadata, None);
+    let reader = read::FileReader::new(arrow_file, metadata, None, None);
     let arrow_schema = reader.schema();
 
     // compare schemas
diff --git a/integration-testing/src/flight_server_scenarios/integration_test.rs b/integration-testing/src/flight_server_scenarios/integration_test.rs
index c224967bfa..9d21cf0c97 100644
--- a/integration-testing/src/flight_server_scenarios/integration_test.rs
+++ b/integration-testing/src/flight_server_scenarios/integration_test.rs
@@ -285,6 +285,7 @@ fn chunk_from_message(
         fields,
         ipc_schema,
         None,
+        None,
         dictionaries,
         arrow_format::ipc::MetadataVersion::V5,
         &mut reader,
diff --git a/tests/it/io/ipc/read/file.rs b/tests/it/io/ipc/read/file.rs
index 022cf4de27..9d21d051f2 100644
--- a/tests/it/io/ipc/read/file.rs
+++ b/tests/it/io/ipc/read/file.rs
@@ -17,7 +17,7 @@ fn test_file(version: &str, file_name: &str) -> Result<()> {
     let (schema, _, batches) = read_gzip_json(version, file_name)?;
 
     let metadata = read_file_metadata(&mut file)?;
-    let reader = FileReader::new(file, metadata, None);
+    let reader = FileReader::new(file, metadata, None, None);
 
     assert_eq!(&schema, reader.schema());
 
@@ -165,14 +165,12 @@ fn test_projection(version: &str, file_name: &str, columns: Vec<usize>) -> Resul
         testdata, version, file_name
     ))?;
 
-    let metadata = read_file_metadata(&mut file)?;
-
-    let (_, _, chunks) = read_gzip_json(version, file_name)?;
+    let (schema, _, chunks) = read_gzip_json(version, file_name)?;
 
     let expected_fields = columns
         .iter()
         .copied()
-        .map(|x| metadata.schema.fields[x].clone())
+        .map(|x| schema.fields[x].clone())
         .collect::<Vec<_>>();
 
     let expected_chunks = chunks.into_iter().map(|chunk| {
@@ -184,7 +182,8 @@ fn test_projection(version: &str, file_name: &str, columns: Vec<usize>) -> Resul
         Chunk::new(columns)
     });
 
-    let reader = FileReader::new(&mut file, metadata, Some(columns.clone()));
+    let metadata = read_file_metadata(&mut file)?;
+    let reader = FileReader::new(&mut file, metadata, Some(columns.clone()), None);
 
     assert_eq!(reader.schema().fields, expected_fields);
 
@@ -210,7 +209,7 @@ fn read_corrupted_ipc(data: Vec<u8>) -> Result<()> {
     let mut file = std::io::Cursor::new(data);
 
     let metadata = read_file_metadata(&mut file)?;
-    let mut reader = FileReader::new(file, metadata, None);
+    let mut reader = FileReader::new(file, metadata, None, None);
 
     reader.try_for_each(|rhs| {
         rhs?;
@@ -241,3 +240,37 @@ fn test_does_not_panic() {
         let _ = read_corrupted_ipc(data);
     }
 }
+
+fn test_limit(version: &str, file_name: &str, limit: usize) -> Result<()> {
+    let testdata = crate::test_util::arrow_test_data();
+    let mut file = File::open(format!(
+        "{}/arrow-ipc-stream/integration/{}/{}.arrow_file",
+        testdata, version, file_name
+    ))?;
+
+    let (schema, _, _) = read_gzip_json(version, file_name)?;
+
+    let metadata = read_file_metadata(&mut file)?;
+    let unlimited_chunk = FileReader::new(&mut file, metadata.clone(), None, None)
+        .next()
+        .unwrap()?;
+    let mut reader = FileReader::new(&mut file, metadata, None, Some(limit));
+
+    assert_eq!(reader.schema(), &schema);
+
+    let chunk = reader.next().unwrap()?;
+    assert_eq!(chunk.len(), unlimited_chunk.len().min(limit));
+
+    Ok(())
+}
+
+#[test]
+fn read_limited() -> Result<()> {
+    test_limit("1.0.0-littleendian", "generated_primitive", 2)?;
+    test_limit("1.0.0-littleendian", "generated_dictionary", 2)?;
+    test_limit("1.0.0-littleendian", "generated_union", 2)?;
+    test_limit("1.0.0-littleendian", "generated_map", 2)?;
+    test_limit("1.0.0-littleendian", "generated_nested_dictionary", 2)?;
+    test_limit("1.0.0-littleendian", "generated_nested", 2)?;
+    Ok(())
+}
diff --git a/tests/it/io/ipc/read_file_async.rs b/tests/it/io/ipc/read_file_async.rs
index af61960460..b8b61853a3 100644
--- a/tests/it/io/ipc/read_file_async.rs
+++ b/tests/it/io/ipc/read_file_async.rs
@@ -17,7 +17,7 @@ async fn test_file(version: &str, file_name: &str) -> Result<()> {
     .compat();
 
     let metadata = read_file_metadata_async(&mut file).await?;
-    let mut reader = FileStream::new(file, metadata, None);
+    let mut reader = FileStream::new(file, metadata, None, None);
 
     // read expected JSON output
     let (schema, ipc_fields, batches) = read_gzip_json(version, file_name)?;
diff --git a/tests/it/io/ipc/write/file.rs b/tests/it/io/ipc/write/file.rs
index 4b261d9773..a899aa459d 100644
--- a/tests/it/io/ipc/write/file.rs
+++ b/tests/it/io/ipc/write/file.rs
@@ -39,7 +39,7 @@ fn round_trip(
     let metadata = read_file_metadata(&mut reader)?;
     let schema = metadata.schema.clone();
 
-    let reader = FileReader::new(reader, metadata, None);
+    let reader = FileReader::new(reader, metadata, None, None);
 
     assert_eq!(schema, expected_schema);
 
@@ -64,7 +64,7 @@ fn test_file(version: &str, file_name: &str, compressed: bool) -> Result<()> {
     let schema = metadata.schema.clone();
     let ipc_fields = metadata.ipc_schema.fields.clone();
 
-    let reader = FileReader::new(reader, metadata, None);
+    let reader = FileReader::new(reader, metadata, None, None);
 
     // read expected JSON output
     let (expected_schema, expected_ipc_fields, expected_batches) =
diff --git a/tests/it/io/ipc/write/file_append.rs b/tests/it/io/ipc/write/file_append.rs
index 6abec8eabc..52edde06d2 100644
--- a/tests/it/io/ipc/write/file_append.rs
+++ b/tests/it/io/ipc/write/file_append.rs
@@ -34,7 +34,7 @@ fn basic() -> Result<()> {
     // read the file again and confirm that it contains both messages
     let metadata = read::read_file_metadata(&mut reader)?;
     assert_eq!(schema, expected_schema);
-    let reader = read::FileReader::new(reader, metadata, None);
+    let reader = read::FileReader::new(reader, metadata, None, None);
 
     let chunks = reader.collect::<Result<Vec<_>>>()?;
 
diff --git a/tests/it/io/ipc/write_file_async.rs b/tests/it/io/ipc/write_file_async.rs
index 25b8164f05..d1963184da 100644
--- a/tests/it/io/ipc/write_file_async.rs
+++ b/tests/it/io/ipc/write_file_async.rs
@@ -38,7 +38,7 @@ async fn test_file(version: &str, file_name: &str) -> Result<()> {
 
     let mut reader = Cursor::new(result);
     let metadata = read::read_file_metadata(&mut reader)?;
-    let reader = read::FileReader::new(reader, metadata, None);
+    let reader = read::FileReader::new(reader, metadata, None, None);
 
     let schema = &reader.metadata().schema;
     let ipc_fields = reader.metadata().ipc_schema.fields.clone();
diff --git a/tests/it/io/parquet/mod.rs b/tests/it/io/parquet/mod.rs
index f43a1736e4..197cf20be5 100644
--- a/tests/it/io/parquet/mod.rs
+++ b/tests/it/io/parquet/mod.rs
@@ -563,7 +563,7 @@ pub fn pyarrow_required_statistics(column: &str) -> Statistics {
 
 pub fn pyarrow_nested_nullable_statistics(column: &str) -> Statistics {
     let new_list = |array: Box<dyn Array>, nullable: bool| {
-        Box::new(ListArray::<i32>::new(
+        ListArray::<i32>::new(
             DataType::List(Box::new(Field::new(
                 "item",
                 array.data_type().clone(),
@@ -572,69 +572,93 @@ pub fn pyarrow_nested_nullable_statistics(column: &str) -> Statistics {
             vec![0, array.len() as i32].into(),
             array,
             None,
-        )) as Box<dyn Array>
+        )
     };
 
     match column {
         "list_int16" => Statistics {
-            distinct_count: Count::Single(UInt64Array::from([None])),
-            null_count: Count::Single(UInt64Array::from([Some(1)])),
-            min_value: new_list(Box::new(Int16Array::from_slice([0])), true),
-            max_value: new_list(Box::new(Int16Array::from_slice([10])), true),
+            distinct_count: Count::List(new_list(UInt64Array::from([None]).boxed(), true)),
+            null_count: Count::List(new_list(UInt64Array::from([Some(1)]).boxed(), true)),
+            min_value: new_list(Box::new(Int16Array::from_slice([0])), true).boxed(),
+            max_value: new_list(Box::new(Int16Array::from_slice([10])), true).boxed(),
         },
         "list_bool" => Statistics {
-            distinct_count: Count::Single(UInt64Array::from([None])),
-            null_count: Count::Single(UInt64Array::from([Some(1)])),
-            min_value: new_list(Box::new(BooleanArray::from_slice([false])), true),
-            max_value: new_list(Box::new(BooleanArray::from_slice([true])), true),
+            distinct_count: Count::List(new_list(UInt64Array::from([None]).boxed(), true)),
+            null_count: Count::List(new_list(UInt64Array::from([Some(1)]).boxed(), true)),
+            min_value: new_list(Box::new(BooleanArray::from_slice([false])), true).boxed(),
+            max_value: new_list(Box::new(BooleanArray::from_slice([true])), true).boxed(),
         },
         "list_utf8" => Statistics {
-            distinct_count: Count::Single(UInt64Array::from([None])),
-            null_count: Count::Single([Some(1)].into()),
-            min_value: new_list(Box::new(Utf8Array::<i32>::from_slice([""])), true),
-            max_value: new_list(Box::new(Utf8Array::<i32>::from_slice(["ccc"])), true),
+            distinct_count: Count::List(new_list(UInt64Array::from([None]).boxed(), true)),
+            null_count: Count::List(new_list(UInt64Array::from([Some(1)]).boxed(), true)),
+            min_value: new_list(Box::new(Utf8Array::<i32>::from_slice([""])), true).boxed(),
+            max_value: new_list(Box::new(Utf8Array::<i32>::from_slice(["ccc"])), true).boxed(),
         },
         "list_large_binary" => Statistics {
-            distinct_count: Count::Single(UInt64Array::from([None])),
-            null_count: Count::Single([Some(1)].into()),
-            min_value: new_list(Box::new(BinaryArray::<i64>::from_slice([b""])), true),
-            max_value: new_list(Box::new(BinaryArray::<i64>::from_slice([b"ccc"])), true),
+            distinct_count: Count::List(new_list(UInt64Array::from([None]).boxed(), true)),
+            null_count: Count::List(new_list(UInt64Array::from([Some(1)]).boxed(), true)),
+            min_value: new_list(Box::new(BinaryArray::<i64>::from_slice([b""])), true).boxed(),
+            max_value: new_list(Box::new(BinaryArray::<i64>::from_slice([b"ccc"])), true).boxed(),
         },
         "list_int64" => Statistics {
-            distinct_count: Count::Single(UInt64Array::from([None])),
-            null_count: Count::Single([Some(1)].into()),
-            min_value: new_list(Box::new(Int64Array::from_slice([0])), true),
-            max_value: new_list(Box::new(Int64Array::from_slice([10])), true),
+            distinct_count: Count::List(new_list(UInt64Array::from([None]).boxed(), true)),
+            null_count: Count::List(new_list(UInt64Array::from([Some(1)]).boxed(), true)),
+            min_value: new_list(Box::new(Int64Array::from_slice([0])), true).boxed(),
+            max_value: new_list(Box::new(Int64Array::from_slice([10])), true).boxed(),
         },
         "list_int64_required" => Statistics {
-            distinct_count: Count::Single(UInt64Array::from([None])),
-            null_count: Count::Single([Some(1)].into()),
-            min_value: new_list(Box::new(Int64Array::from_slice([0])), false),
-            max_value: new_list(Box::new(Int64Array::from_slice([10])), false),
+            distinct_count: Count::List(new_list(UInt64Array::from([None]).boxed(), true)),
+            null_count: Count::List(new_list(UInt64Array::from([Some(1)]).boxed(), true)),
+            min_value: new_list(Box::new(Int64Array::from_slice([0])), false).boxed(),
+            max_value: new_list(Box::new(Int64Array::from_slice([10])), false).boxed(),
         },
         "list_int64_required_required" | "list_int64_optional_required" => Statistics {
-            distinct_count: Count::Single(UInt64Array::from([None])),
-            null_count: Count::Single([Some(0)].into()),
-            min_value: new_list(Box::new(Int64Array::from_slice([0])), false),
-            max_value: new_list(Box::new(Int64Array::from_slice([10])), false),
+            distinct_count: Count::List(new_list(UInt64Array::from([None]).boxed(), false)),
+            null_count: Count::List(new_list(UInt64Array::from([Some(0)]).boxed(), false)),
+            min_value: new_list(Box::new(Int64Array::from_slice([0])), false).boxed(),
+            max_value: new_list(Box::new(Int64Array::from_slice([10])), false).boxed(),
         },
         "list_nested_i64" => Statistics {
-            distinct_count: Count::Single(UInt64Array::from([None])),
-            null_count: Count::Single([Some(2)].into()),
-            min_value: new_list(new_list(Box::new(Int64Array::from_slice([0])), true), true),
-            max_value: new_list(new_list(Box::new(Int64Array::from_slice([10])), true), true),
+            distinct_count: Count::List(new_list(UInt64Array::from([None]).boxed(), true)),
+            null_count: Count::List(new_list(UInt64Array::from([Some(2)]).boxed(), true)),
+            min_value: new_list(
+                new_list(Box::new(Int64Array::from_slice([0])), true).boxed(),
+                true,
+            )
+            .boxed(),
+            max_value: new_list(
+                new_list(Box::new(Int64Array::from_slice([10])), true).boxed(),
+                true,
+            )
+            .boxed(),
         },
         "list_nested_inner_required_required_i64" => Statistics {
             distinct_count: Count::Single(UInt64Array::from([None])),
             null_count: Count::Single([Some(0)].into()),
-            min_value: new_list(new_list(Box::new(Int64Array::from_slice([0])), true), true),
-            max_value: new_list(new_list(Box::new(Int64Array::from_slice([10])), true), true),
+            min_value: new_list(
+                new_list(Box::new(Int64Array::from_slice([0])), true).boxed(),
+                true,
+            )
+            .boxed(),
+            max_value: new_list(
+                new_list(Box::new(Int64Array::from_slice([10])), true).boxed(),
+                true,
+            )
+            .boxed(),
         },
         "list_nested_inner_required_i64" => Statistics {
             distinct_count: Count::Single(UInt64Array::from([None])),
             null_count: Count::Single([Some(0)].into()),
-            min_value: new_list(new_list(Box::new(Int64Array::from_slice([0])), true), true),
-            max_value: new_list(new_list(Box::new(Int64Array::from_slice([10])), true), true),
+            min_value: new_list(
+                new_list(Box::new(Int64Array::from_slice([0])), true).boxed(),
+                true,
+            )
+            .boxed(),
+            max_value: new_list(
+                new_list(Box::new(Int64Array::from_slice([10])), true).boxed(),
+                true,
+            )
+            .boxed(),
         },
         other => todo!("{}", other),
     }
@@ -642,7 +666,7 @@ pub fn pyarrow_nested_nullable_statistics(column: &str) -> Statistics {
 
 pub fn pyarrow_nested_edge_statistics(column: &str) -> Statistics {
     let new_list = |array: Box<dyn Array>| {
-        Box::new(ListArray::<i32>::new(
+        ListArray::<i32>::new(
             DataType::List(Box::new(Field::new(
                 "item",
                 array.data_type().clone(),
@@ -651,21 +675,21 @@ pub fn pyarrow_nested_edge_statistics(column: &str) -> Statistics {
             vec![0, array.len() as i32].into(),
             array,
             None,
-        ))
+        )
     };
 
     match column {
         "simple" => Statistics {
-            distinct_count: Count::Single(UInt64Array::from([None])),
-            null_count: Count::Single(UInt64Array::from([Some(0)])),
-            min_value: new_list(Box::new(Int64Array::from([Some(0)]))),
-            max_value: new_list(Box::new(Int64Array::from([Some(1)]))),
+            distinct_count: Count::List(new_list(UInt64Array::from([None]).boxed())),
+            null_count: Count::List(new_list(UInt64Array::from([Some(0)]).boxed())),
+            min_value: new_list(Box::new(Int64Array::from([Some(0)]))).boxed(),
+            max_value: new_list(Box::new(Int64Array::from([Some(1)]))).boxed(),
         },
         "null" => Statistics {
-            distinct_count: Count::Single(UInt64Array::from([None])),
-            null_count: Count::Single(UInt64Array::from([Some(1)])),
-            min_value: new_list(Box::new(Int64Array::from([None]))),
-            max_value: new_list(Box::new(Int64Array::from([None]))),
+            distinct_count: Count::List(new_list(UInt64Array::from([None]).boxed())),
+            null_count: Count::List(new_list(UInt64Array::from([Some(1)]).boxed())),
+            min_value: new_list(Box::new(Int64Array::from([None]))).boxed(),
+            max_value: new_list(Box::new(Int64Array::from([None]))).boxed(),
         },
         _ => unreachable!(),
     }