Repository: jorgecarleitao/arrow2
Pull request #1137 (2022-07-03 17:12:39)

Improved dictionary invariants
This PR changes the internal invariant from `DictionaryArray` to only contain keys that:
* can be casted to `usize`
* the maximum value is smaller than the values' length

This allows removing bound checks when iterating over the values via its keys.

# Backward incompatible changes

* `DictionaryArray::from_data` was replaced by `try_new`, `try_new_unchecked` and `try_from_keys`
* `DictionaryKey` now only implements `NativeType + TryFrom<usize> + TryInto<usize>`


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

```diff
@@            Coverage Diff             @@
##             main    #1137      +/-   ##
==========================================
- Coverage   83.49%   83.41%   -0.08%     
==========================================
  Files         366      366              
  Lines       35635    35832     +197     
==========================================
+ Hits        29752    29889     +137     
- Misses       5883     5943      +60     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/array/dictionary/iterator.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2RpY3Rpb25hcnkvaXRlcmF0b3IucnM=) | `55.55% <ø> (-2.78%)` | :arrow_down: |
| [src/array/equal/dictionary.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2VxdWFsL2RpY3Rpb25hcnkucnM=) | `100.00% <ø> (ø)` | |
| [src/array/equal/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2VxdWFsL21vZC5ycw==) | `80.00% <0.00%> (-3.59%)` | :arrow_down: |
| [src/compute/arithmetics/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2NvbXB1dGUvYXJpdGhtZXRpY3MvbW9kLnJz) | `74.10% <ø> (ø)` | |
| [src/io/parquet/read/statistics/dictionary.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvcmVhZC9zdGF0aXN0aWNzL2RpY3Rpb25hcnkucnM=) | `48.57% <ø> (ø)` | |
| [src/compute/cast/dictionary\_to.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2NvbXB1dGUvY2FzdC9kaWN0aW9uYXJ5X3RvLnJz) | `22.41% <7.14%> (-4.40%)` | :arrow_down: |
| [src/scalar/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL3NjYWxhci9tb2QucnM=) | `69.11% <38.09%> (-13.87%)` | :arrow_down: |
| [src/io/avro/read/nested.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2F2cm8vcmVhZC9uZXN0ZWQucnM=) | `64.53% <50.00%> (-0.60%)` | :arrow_down: |
| [src/io/parquet/read/deserialize/dictionary.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvcmVhZC9kZXNlcmlhbGl6ZS9kaWN0aW9uYXJ5LnJz) | `79.06% <65.62%> (-2.69%)` | :arrow_down: |
| [src/array/growable/dictionary.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2dyb3dhYmxlL2RpY3Rpb25hcnkucnM=) | `77.31% <72.22%> (-0.21%)` | :arrow_down: |
| ... and [25 more](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1137/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | |

------

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


Patch:
diff --git a/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/array/dictionary/ffi.rs b/src/array/dictionary/ffi.rs
index 6bc467630a..1e06f81e6e 100644
--- a/src/array/dictionary/ffi.rs
+++ b/src/array/dictionary/ffi.rs
@@ -1,6 +1,6 @@
 use crate::{
     array::{FromFfi, PrimitiveArray, ToFfi},
-    error::Result,
+    error::Error,
     ffi,
 };
 
@@ -25,16 +25,20 @@ unsafe impl<K: DictionaryKey> ToFfi for DictionaryArray<K> {
 }
 
 impl<K: DictionaryKey, A: ffi::ArrowArrayRef> FromFfi<A> for DictionaryArray<K> {
-    unsafe fn try_from_ffi(array: A) -> Result<Self> {
+    unsafe fn try_from_ffi(array: A) -> Result<Self, Error> {
         // keys: similar to PrimitiveArray, but the datatype is the inner one
         let validity = unsafe { array.validity() }?;
         let values = unsafe { array.buffer::<K>(1) }?;
 
-        let data_type = K::PRIMITIVE.into();
-        let keys = PrimitiveArray::<K>::try_new(data_type, values, validity)?;
-        let values = array.dictionary()?.unwrap();
+        let data_type = array.data_type().clone();
+
+        let keys = PrimitiveArray::<K>::try_new(K::PRIMITIVE.into(), values, validity)?;
+        let values = array
+            .dictionary()?
+            .ok_or_else(|| Error::oos("Dictionary Array must contain a dictionary in ffi"))?;
         let values = ffi::try_from(values)?;
 
-        Ok(DictionaryArray::<K>::from_data(keys, values))
+        // the assumption of this trait
+        DictionaryArray::<K>::try_new_unchecked(data_type, keys, values)
     }
 }
diff --git a/src/array/dictionary/fmt.rs b/src/array/dictionary/fmt.rs
index e3e3b475d8..f5f7662400 100644
--- a/src/array/dictionary/fmt.rs
+++ b/src/array/dictionary/fmt.rs
@@ -15,7 +15,7 @@ pub fn write_value<K: DictionaryKey, W: Write>(
     let values = array.values();
 
     if keys.is_valid(index) {
-        let key = keys.value(index).to_usize().unwrap();
+        let key = array.key_value(index);
         get_display(values.as_ref(), null)(f, key)
     } else {
         write!(f, "{}", null)
diff --git a/src/array/dictionary/iterator.rs b/src/array/dictionary/iterator.rs
index 21b1cc7bad..0249a1b589 100644
--- a/src/array/dictionary/iterator.rs
+++ b/src/array/dictionary/iterator.rs
@@ -1,4 +1,4 @@
-use crate::bitmap::utils::{zip_validity, ZipValidity};
+use crate::bitmap::utils::ZipValidity;
 use crate::scalar::Scalar;
 use crate::trusted_len::TrustedLen;
 
@@ -66,18 +66,3 @@ impl<'a, K: DictionaryKey> IntoIterator for &'a DictionaryArray<K> {
         self.iter()
     }
 }
-
-impl<'a, K: DictionaryKey> DictionaryArray<K> {
-    /// Returns an iterator of `Option<Box<dyn Array>>`
-    pub fn iter(&'a self) -> ZipIter<'a, K> {
-        zip_validity(
-            DictionaryValuesIter::new(self),
-            self.keys.validity().as_ref().map(|x| x.iter()),
-        )
-    }
-
-    /// Returns an iterator of `Box<dyn Array>`
-    pub fn values_iter(&'a self) -> ValuesIter<'a, K> {
-        DictionaryValuesIter::new(self)
-    }
-}
diff --git a/src/array/dictionary/mod.rs b/src/array/dictionary/mod.rs
index 0c9d926737..0d5b4502eb 100644
--- a/src/array/dictionary/mod.rs
+++ b/src/array/dictionary/mod.rs
@@ -1,7 +1,14 @@
+use std::hint::unreachable_unchecked;
+
 use crate::{
-    bitmap::Bitmap,
+    bitmap::{
+        utils::{zip_validity, ZipValidity},
+        Bitmap,
+    },
     datatypes::{DataType, IntegerType},
+    error::Error,
     scalar::{new_scalar, Scalar},
+    trusted_len::TrustedLen,
     types::NativeType,
 };
 
@@ -13,12 +20,23 @@ pub use iterator::*;
 pub use mutable::*;
 
 use super::{new_empty_array, primitive::PrimitiveArray, Array};
-use crate::scalar::NullScalar;
+use super::{new_null_array, specification::check_indexes};
 
 /// Trait denoting [`NativeType`]s that can be used as keys of a dictionary.
-pub trait DictionaryKey: NativeType + num_traits::NumCast + num_traits::FromPrimitive {
+pub trait DictionaryKey: NativeType + TryInto<usize> + TryFrom<usize> {
     /// The corresponding [`IntegerType`] of this key
     const KEY_TYPE: IntegerType;
+
+    /// Represents this key as a `usize`.
+    /// # Safety
+    /// The caller _must_ have checked that the value can be casted to `usize`.
+    #[inline]
+    unsafe fn as_usize(self) -> usize {
+        match self.try_into() {
+            Ok(v) => v,
+            Err(_) => unreachable_unchecked(),
+        }
+    }
 }
 
 impl DictionaryKey for i8 {
@@ -46,8 +64,13 @@ impl DictionaryKey for u64 {
     const KEY_TYPE: IntegerType = IntegerType::UInt64;
 }
 
-/// An [`Array`] whose values are encoded by keys. This [`Array`] is useful when the cardinality of
+/// An [`Array`] whose values are stored as indices. This [`Array`] is useful when the cardinality of
 /// values is low compared to the length of the [`Array`].
+///
+/// # Safety
+/// This struct guarantees that each item of [`DictionaryArray::keys`] is castable to `usize` and
+/// its value is smaller than [`DictionaryArray::values`]`.len()`. In other words, you can safely
+/// use `unchecked` calls to retrive the values
 #[derive(Clone)]
 pub struct DictionaryArray<K: DictionaryKey> {
     data_type: DataType,
@@ -55,38 +78,152 @@ pub struct DictionaryArray<K: DictionaryKey> {
     values: Box<dyn Array>,
 }
 
+fn check_data_type(
+    key_type: IntegerType,
+    data_type: &DataType,
+    values_data_type: &DataType,
+) -> Result<(), Error> {
+    if let DataType::Dictionary(key, value, _) = data_type.to_logical_type() {
+        if *key != key_type {
+            return Err(Error::oos(
+                "DictionaryArray must be initialized with a DataType::Dictionary whose integer is compatible to its keys",
+            ));
+        }
+        if value.as_ref().to_logical_type() != values_data_type.to_logical_type() {
+            return Err(Error::oos(
+                "DictionaryArray must be initialized with a DataType::Dictionary whose value is equal to its values",
+            ));
+        }
+    } else {
+        return Err(Error::oos(
+            "DictionaryArray must be initialized with logical DataType::Dictionary",
+        ));
+    }
+    Ok(())
+}
+
 impl<K: DictionaryKey> DictionaryArray<K> {
+    /// Returns a new [`DictionaryArray`].
+    /// # Implementation
+    /// This function is `O(N)` where `N` is the length of keys
+    /// # Errors
+    /// This function errors iff
+    /// * the `data_type`'s logical type is not a `DictionaryArray`
+    /// * the `data_type`'s keys is not compatible with `keys`
+    /// * the `data_type`'s values's data_type is not equal with `values.data_type()`
+    /// * any of the keys's values is not represented in `usize` or is `>= values.len()`
+    pub fn try_new(
+        data_type: DataType,
+        keys: PrimitiveArray<K>,
+        values: Box<dyn Array>,
+    ) -> Result<Self, Error> {
+        check_data_type(K::KEY_TYPE, &data_type, values.data_type())?;
+
+        check_indexes(keys.values(), values.len())?;
+
+        Ok(Self {
+            data_type,
+            keys,
+            values,
+        })
+    }
+
+    /// Returns a new [`DictionaryArray`].
+    /// # Implementation
+    /// This function is `O(N)` where `N` is the length of keys
+    /// # Errors
+    /// This function errors iff
+    /// * any of the keys's values is not represented in `usize` or is `>= values.len()`
+    pub fn try_from_keys(keys: PrimitiveArray<K>, values: Box<dyn Array>) -> Result<Self, Error> {
+        let data_type = Self::default_data_type(values.data_type().clone());
+        Self::try_new(data_type, keys, values)
+    }
+
+    /// Returns a new [`DictionaryArray`].
+    /// # Errors
+    /// This function errors iff
+    /// * the `data_type`'s logical type is not a `DictionaryArray`
+    /// * the `data_type`'s keys is not compatible with `keys`
+    /// * the `data_type`'s values's data_type is not equal with `values.data_type()`
+    /// # Safety
+    /// The caller must ensure that every keys's values is represented in `usize` and is `< values.len()`
+    pub unsafe fn try_new_unchecked(
+        data_type: DataType,
+        keys: PrimitiveArray<K>,
+        values: Box<dyn Array>,
+    ) -> Result<Self, Error> {
+        check_data_type(K::KEY_TYPE, &data_type, values.data_type())?;
+
+        Ok(Self {
+            data_type,
+            keys,
+            values,
+        })
+    }
+
     /// Returns a new empty [`DictionaryArray`].
     pub fn new_empty(data_type: DataType) -> Self {
-        let values = Self::get_child(&data_type);
+        let values = Self::try_get_child(&data_type).unwrap();
         let values = new_empty_array(values.clone());
-        let data_type = K::PRIMITIVE.into();
-        Self::from_data(PrimitiveArray::<K>::new_empty(data_type), values)
+        Self::try_new(
+            data_type,
+            PrimitiveArray::<K>::new_empty(K::PRIMITIVE.into()),
+            values,
+        )
+        .unwrap()
     }
 
     /// Returns an [`DictionaryArray`] whose all elements are null
     #[inline]
     pub fn new_null(data_type: DataType, length: usize) -> Self {
-        let values = Self::get_child(&data_type);
-        let data_type = K::PRIMITIVE.into();
-        Self::from_data(
-            PrimitiveArray::<K>::new_null(data_type, length),
-            new_empty_array(values.clone()),
+        let values = Self::try_get_child(&data_type).unwrap();
+        let values = new_null_array(values.clone(), 1);
+        Self::try_new(
+            data_type,
+            PrimitiveArray::<K>::new_null(K::PRIMITIVE.into(), length),
+            values,
         )
+        .unwrap()
     }
 
-    /// The canonical method to create a new [`DictionaryArray`].
-    pub fn from_data(keys: PrimitiveArray<K>, values: Box<dyn Array>) -> Self {
-        let data_type =
-            DataType::Dictionary(K::KEY_TYPE, Box::new(values.data_type().clone()), false);
+    /// Returns an iterator of [`Option<Box<dyn Scalar>>`].
+    /// # Implementation
+    /// This function will allocate a new [`Scalar`] per item and is usually not performant.
+    /// Consider calling `keys_iter` and `values`, downcasting `values`, and iterating over that.
+    pub fn iter(&self) -> ZipValidity<Box<dyn Scalar>, DictionaryValuesIter<K>> {
+        zip_validity(
+            DictionaryValuesIter::new(self),
+            self.keys.validity().as_ref().map(|x| x.iter()),
+        )
+    }
 
-        Self {
-            data_type,
-            keys,
-            values,
+    /// Returns an iterator of [`Box<dyn Scalar>`]
+    /// # Implementation
+    /// This function will allocate a new [`Scalar`] per item and is usually not performant.
+    /// Consider calling `keys_iter` and `values`, downcasting `values`, and iterating over that.
+    pub fn values_iter(&self) -> DictionaryValuesIter<K> {
+        DictionaryValuesIter::new(self)
+    }
+
+    /// Returns the [`DataType`] of this [`DictionaryArray`]
+    #[inline]
+    pub fn data_type(&self) -> &DataType {
+        &self.data_type
+    }
+
+    /// Returns whether the values of this [`DictionaryArray`] are ordered
+    #[inline]
+    pub fn is_ordered(&self) -> bool {
+        match self.data_type.to_logical_type() {
+            DataType::Dictionary(_, _, is_ordered) => *is_ordered,
+            _ => unreachable!(),
         }
     }
 
+    pub(crate) fn default_data_type(values_datatype: DataType) -> DataType {
+        DataType::Dictionary(K::KEY_TYPE, Box::new(values_datatype), false)
+    }
+
     /// Creates a new [`DictionaryArray`] by slicing the existing [`DictionaryArray`].
     /// # Panics
     /// iff `offset + length > self.len()`.
@@ -124,10 +261,7 @@ impl<K: DictionaryKey> DictionaryArray<K> {
     pub fn set_validity(&mut self, validity: Option<Bitmap>) {
         self.keys.set_validity(validity);
     }
-}
 
-// accessors
-impl<K: DictionaryKey> DictionaryArray<K> {
     /// Returns the length of this array
     #[inline]
     pub fn len(&self) -> usize {
@@ -147,6 +281,29 @@ impl<K: DictionaryKey> DictionaryArray<K> {
         &self.keys
     }
 
+    /// Returns an iterator of the keys' values of the [`DictionaryArray`] as `usize`
+    #[inline]
+    pub fn keys_values_iter(&self) -> impl TrustedLen<Item = usize> + Clone + '_ {
+        // safety - invariant of the struct
+        self.keys.values_iter().map(|x| unsafe { x.as_usize() })
+    }
+
+    /// Returns an iterator of the keys' of the [`DictionaryArray`] as `usize`
+    #[inline]
+    pub fn keys_iter(&self) -> impl TrustedLen<Item = Option<usize>> + Clone + '_ {
+        // safety - invariant of the struct
+        self.keys.iter().map(|x| x.map(|x| unsafe { x.as_usize() }))
+    }
+
+    /// Returns the keys' value of the [`DictionaryArray`] as `usize`
+    /// # Panics
+    /// This function panics iff `index >= self.len()`
+    #[inline]
+    pub fn key_value(&self, index: usize) -> usize {
+        // safety - invariant of the struct
+        unsafe { self.keys.values()[index].as_usize() }
+    }
+
     /// Returns the values of the [`DictionaryArray`].
     #[inline]
     pub fn values(&self) -> &Box<dyn Array> {
@@ -154,14 +311,16 @@ impl<K: DictionaryKey> DictionaryArray<K> {
     }
 
     /// Returns the value of the [`DictionaryArray`] at position `i`.
+    /// # Implementation
+    /// This function will allocate a new [`Scalar`] and is usually not performant.
+    /// Consider calling `keys` and `values`, downcasting `values`, and iterating over that.
+    /// # Panic
+    /// This function panics iff `index >= self.len()`
     #[inline]
     pub fn value(&self, index: usize) -> Box<dyn Scalar> {
-        if self.keys.is_null(index) {
-            Box::new(NullScalar::new())
-        } else {
-            let index = self.keys.value(index).to_usize().unwrap();
-            new_scalar(self.values.as_ref(), index)
-        }
+        // safety - invariant of this struct
+        let index = unsafe { self.keys.value(index).as_usize() };
+        new_scalar(self.values.as_ref(), index)
     }
 
     /// Boxes self into a [`Box<dyn Array>`].
@@ -173,15 +332,16 @@ impl<K: DictionaryKey> DictionaryArray<K> {
     pub fn arced(self) -> std::sync::Arc<dyn Array> {
         std::sync::Arc::new(self)
     }
-}
 
-impl<K: DictionaryKey> DictionaryArray<K> {
-    pub(crate) fn get_child(data_type: &DataType) -> &DataType {
-        match data_type {
+    pub(crate) fn try_get_child(data_type: &DataType) -> Result<&DataType, Error> {
+        Ok(match data_type.to_logical_type() {
             DataType::Dictionary(_, values, _) => values.as_ref(),
-            DataType::Extension(_, inner, _) => Self::get_child(inner),
-            _ => panic!("DictionaryArray must be initialized with DataType::Dictionary"),
-        }
+            _ => {
+                return Err(Error::oos(
+                    "Dictionaries must be initialized with DataType::Dictionary",
+                ))
+            }
+        })
     }
 }
 
@@ -213,12 +373,15 @@ impl<K: DictionaryKey> Array for DictionaryArray<K> {
     fn slice(&self, offset: usize, length: usize) -> Box<dyn Array> {
         Box::new(self.slice(offset, length))
     }
+
     unsafe fn slice_unchecked(&self, offset: usize, length: usize) -> Box<dyn Array> {
         Box::new(self.slice_unchecked(offset, length))
     }
+
     fn with_validity(&self, validity: Option<Bitmap>) -> Box<dyn Array> {
         Box::new(self.clone().with_validity(validity))
     }
+
     fn to_boxed(&self) -> Box<dyn Array> {
         Box::new(self.clone())
     }
diff --git a/src/array/dictionary/mutable.rs b/src/array/dictionary/mutable.rs
index ba0d393f97..c5ca7efb31 100644
--- a/src/array/dictionary/mutable.rs
+++ b/src/array/dictionary/mutable.rs
@@ -32,12 +32,21 @@ pub struct MutableDictionaryArray<K: DictionaryKey, M: MutableArray> {
     data_type: DataType,
     keys: MutablePrimitiveArray<K>,
     map: HashedMap<u64, K>,
+    // invariant: `keys.len() <= values.len()`
     values: M,
 }
 
 impl<K: DictionaryKey, M: MutableArray> From<MutableDictionaryArray<K, M>> for DictionaryArray<K> {
     fn from(mut other: MutableDictionaryArray<K, M>) -> Self {
-        DictionaryArray::<K>::from_data(other.keys.into(), other.values.as_box())
+        // Safety - the invariant of this struct ensures that this is up-held
+        unsafe {
+            DictionaryArray::<K>::try_new_unchecked(
+                other.data_type,
+                other.keys.into(),
+                other.values.as_box(),
+            )
+            .unwrap()
+        }
     }
 }
 
@@ -91,7 +100,7 @@ impl<K: DictionaryKey, M: MutableArray> MutableDictionaryArray<K, M> {
                 Ok(false)
             }
             None => {
-                let key = K::from_usize(self.map.len()).ok_or(Error::Overflow)?;
+                let key = K::try_from(self.map.len()).map_err(|_| Error::Overflow)?;
                 self.map.insert(hash, key);
                 self.keys.push(Some(key));
                 Ok(true)
@@ -105,7 +114,7 @@ impl<K: DictionaryKey, M: MutableArray> MutableDictionaryArray<K, M> {
     }
 
     /// returns a mutable reference to the inner values.
-    pub fn mut_values(&mut self) -> &mut M {
+    fn mut_values(&mut self) -> &mut M {
         &mut self.values
     }
 
@@ -141,6 +150,18 @@ impl<K: DictionaryKey, M: MutableArray> MutableDictionaryArray<K, M> {
     pub fn keys(&self) -> &MutablePrimitiveArray<K> {
         &self.keys
     }
+
+    fn take_into(&mut self) -> DictionaryArray<K> {
+        // Safety - the invariant of this struct ensures that this is up-held
+        unsafe {
+            DictionaryArray::<K>::try_new(
+                self.data_type.clone(),
+                std::mem::take(&mut self.keys).into(),
+                self.values.as_box(),
+            )
+            .unwrap()
+        }
+    }
 }
 
 impl<K: DictionaryKey, M: 'static + MutableArray> MutableArray for MutableDictionaryArray<K, M> {
@@ -153,17 +174,11 @@ impl<K: DictionaryKey, M: 'static + MutableArray> MutableArray for MutableDictio
     }
 
     fn as_box(&mut self) -> Box<dyn Array> {
-        Box::new(DictionaryArray::<K>::from_data(
-            std::mem::take(&mut self.keys).into(),
-            self.values.as_box(),
-        ))
+        Box::new(self.take_into())
     }
 
     fn as_arc(&mut self) -> Arc<dyn Array> {
-        Arc::new(DictionaryArray::<K>::from_data(
-            std::mem::take(&mut self.keys).into(),
-            self.values.as_box(),
-        ))
+        Arc::new(self.take_into())
     }
 
     fn data_type(&self) -> &DataType {
diff --git a/src/array/equal/dictionary.rs b/src/array/equal/dictionary.rs
index 8c879ff837..d65634095f 100644
--- a/src/array/equal/dictionary.rs
+++ b/src/array/equal/dictionary.rs
@@ -1,4 +1,4 @@
-use crate::array::{Array, DictionaryArray, DictionaryKey};
+use crate::array::{DictionaryArray, DictionaryKey};
 
 pub(super) fn equal<K: DictionaryKey>(lhs: &DictionaryArray<K>, rhs: &DictionaryArray<K>) -> bool {
     if !(lhs.data_type() == rhs.data_type() && lhs.len() == rhs.len()) {
diff --git a/src/array/equal/mod.rs b/src/array/equal/mod.rs
index 737edb35e7..aa2ea60288 100644
--- a/src/array/equal/mod.rs
+++ b/src/array/equal/mod.rs
@@ -87,6 +87,12 @@ impl<O: Offset> PartialEq<&dyn Array> for Utf8Array<O> {
     }
 }
 
+impl<O: Offset> PartialEq<Utf8Array<O>> for &dyn Array {
+    fn eq(&self, other: &Utf8Array<O>) -> bool {
+        equal(*self, other)
+    }
+}
+
 impl<O: Offset> PartialEq<BinaryArray<O>> for BinaryArray<O> {
     fn eq(&self, other: &Self) -> bool {
         binary::equal(self, other)
@@ -99,6 +105,12 @@ impl<O: Offset> PartialEq<&dyn Array> for BinaryArray<O> {
     }
 }
 
+impl<O: Offset> PartialEq<BinaryArray<O>> for &dyn Array {
+    fn eq(&self, other: &BinaryArray<O>) -> bool {
+        equal(*self, other)
+    }
+}
+
 impl PartialEq<FixedSizeBinaryArray> for FixedSizeBinaryArray {
     fn eq(&self, other: &Self) -> bool {
         fixed_size_binary::equal(self, other)
diff --git a/src/array/growable/dictionary.rs b/src/array/growable/dictionary.rs
index 908a1394fa..0b3c432525 100644
--- a/src/array/growable/dictionary.rs
+++ b/src/array/growable/dictionary.rs
@@ -3,6 +3,7 @@ use std::sync::Arc;
 use crate::{
     array::{Array, DictionaryArray, DictionaryKey, PrimitiveArray},
     bitmap::MutableBitmap,
+    datatypes::DataType,
 };
 
 use super::{
@@ -16,6 +17,7 @@ use super::{
 /// This growable does not perform collision checks and instead concatenates
 /// the values of each [`DictionaryArray`] one after the other.
 pub struct GrowableDictionary<'a, K: DictionaryKey> {
+    data_type: DataType,
     keys_values: Vec<&'a [K]>,
     key_values: Vec<K>,
     key_validity: MutableBitmap,
@@ -44,6 +46,8 @@ impl<'a, T: DictionaryKey> GrowableDictionary<'a, T> {
     /// # Panics
     /// If `arrays` is empty.
     pub fn new(arrays: &[&'a DictionaryArray<T>], mut use_validity: bool, capacity: usize) -> Self {
+        let data_type = arrays[0].data_type().clone();
+
         // if any of the arrays has nulls, insertions from any array requires setting bits
         // as there is at least one array with nulls.
         if arrays.iter().any(|array| array.null_count() > 0) {
@@ -69,6 +73,7 @@ impl<'a, T: DictionaryKey> GrowableDictionary<'a, T> {
         let (values, offsets) = concatenate_values(&arrays_keys, &arrays_values, capacity);
 
         Self {
+            data_type,
             offsets,
             values,
             keys_values,
@@ -83,10 +88,11 @@ impl<'a, T: DictionaryKey> GrowableDictionary<'a, T> {
         let validity = std::mem::take(&mut self.key_validity);
         let key_values = std::mem::take(&mut self.key_values);
 
-        let data_type = T::PRIMITIVE.into();
-        let keys = PrimitiveArray::<T>::from_data(data_type, key_values.into(), validity.into());
+        let keys =
+            PrimitiveArray::<T>::try_new(T::PRIMITIVE.into(), key_values.into(), validity.into())
+                .unwrap();
 
-        DictionaryArray::<T>::from_data(keys, self.values.clone())
+        DictionaryArray::<T>::try_new(self.data_type.clone(), keys, self.values.clone()).unwrap()
     }
 }
 
@@ -101,7 +107,17 @@ impl<'a, T: DictionaryKey> Growable<'a> for GrowableDictionary<'a, T> {
             values
                 .iter()
                 // `.unwrap_or(0)` because this operation does not check for null values, which may contain any key.
-                .map(|x| T::from_usize(offset + x.to_usize().unwrap_or(0)).unwrap()),
+                .map(|x| {
+                    let x: usize = offset + (*x).try_into().unwrap_or(0);
+                    let x: T = match x.try_into() {
+                        Ok(key) => key,
+                        // todo: convert this to an error.
+                        Err(_) => {
+                            panic!("The maximum key is too small")
+                        }
+                    };
+                    x
+                }),
         );
     }
 
@@ -127,12 +143,10 @@ impl<'a, T: DictionaryKey> From<GrowableDictionary<'a, T>> for DictionaryArray<T
     #[inline]
     fn from(val: GrowableDictionary<'a, T>) -> Self {
         let data_type = T::PRIMITIVE.into();
-        let keys = PrimitiveArray::<T>::from_data(
-            data_type,
-            val.key_values.into(),
-            val.key_validity.into(),
-        );
+        let keys =
+            PrimitiveArray::<T>::try_new(data_type, val.key_values.into(), val.key_validity.into())
+                .unwrap();
 
-        DictionaryArray::<T>::from_data(keys, val.values)
+        DictionaryArray::<T>::try_new(val.data_type.clone(), keys, val.values).unwrap()
     }
 }
diff --git a/src/array/ord.rs b/src/array/ord.rs
index 8fe3adec31..639317165a 100644
--- a/src/array/ord.rs
+++ b/src/array/ord.rs
@@ -141,8 +141,9 @@ where
     let comparator = build_compare(left.values().as_ref(), right.values().as_ref())?;
 
     Ok(Box::new(move |i: usize, j: usize| {
-        let key_left = left_keys[i].to_usize().unwrap();
-        let key_right = right_keys[j].to_usize().unwrap();
+        // safety: all dictionaries keys are guaranteed to be castable to usize
+        let key_left = unsafe { left_keys[i].as_usize() };
+        let key_right = unsafe { right_keys[j].as_usize() };
         (comparator)(key_left, key_right)
     }))
 }
diff --git a/src/array/specification.rs b/src/array/specification.rs
index 0ede258232..8c35194f96 100644
--- a/src/array/specification.rs
+++ b/src/array/specification.rs
@@ -97,6 +97,22 @@ pub fn try_check_offsets<O: Offset>(offsets: &[O], values_len: usize) -> Result<
     }
 }
 
+pub fn check_indexes<K>(keys: &[K], len: usize) -> Result<()>
+where
+    K: Copy + TryInto<usize>,
+{
+    keys.iter().try_for_each(|key| {
+        let key: usize = (*key)
+            .try_into()
+            .map_err(|_| Error::oos("The dictionary key must fit in a `usize`"))?;
+        if key >= len {
+            Err(Error::oos("The dictionary key must be smaller"))
+        } else {
+            Ok(())
+        }
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use proptest::prelude::*;
diff --git a/src/compute/arithmetics/mod.rs b/src/compute/arithmetics/mod.rs
index d0ece21795..b3a52cd7e1 100644
--- a/src/compute/arithmetics/mod.rs
+++ b/src/compute/arithmetics/mod.rs
@@ -434,9 +434,12 @@ pub fn neg(array: &dyn Array) -> Box<dyn Array> {
         Dictionary(key) => match_integer_type!(key, |$T| {
             let array = array.as_any().downcast_ref::<DictionaryArray<$T>>().unwrap();
 
-            let values = neg(array.values().as_ref()).into();
+            let values = neg(array.values().as_ref());
 
-            Box::new(DictionaryArray::<$T>::from_data(array.keys().clone(), values)) as Box<dyn Array>
+            // safety - this operation only applies to values and thus preserves the dictionary's invariant
+            unsafe{
+                DictionaryArray::<$T>::try_new_unchecked(array.data_type().clone(), array.keys().clone(), values).unwrap().boxed()
+            }
         }),
         _ => todo!(),
     }
diff --git a/src/compute/cast/dictionary_to.rs b/src/compute/cast/dictionary_to.rs
index 16b1c06f10..01d3325cf9 100644
--- a/src/compute/cast/dictionary_to.rs
+++ b/src/compute/cast/dictionary_to.rs
@@ -15,9 +15,7 @@ macro_rules! key_cast {
         if cast_keys.null_count() > $keys.null_count() {
             return Err(Error::Overflow);
         }
-        Ok(Box::new(DictionaryArray::<$to_type>::from_data(
-            cast_keys, $values,
-        )))
+        DictionaryArray::try_new($array.data_type().clone(), $keys.clone(), $values.clone())
     }};
 }
 
@@ -31,9 +29,14 @@ pub fn dictionary_to_dictionary_values<K: DictionaryKey>(
 ) -> Result<DictionaryArray<K>> {
     let keys = from.keys();
     let values = from.values();
+    let length = values.len();
 
     let values = cast(values.as_ref(), values_type, CastOptions::default())?;
-    Ok(DictionaryArray::from_data(keys.clone(), values))
+
+    assert_eq!(values.len(), length); // this is guaranteed by `cast`
+    unsafe {
+        DictionaryArray::try_new_unchecked(from.data_type().clone(), keys.clone(), values.clone())
+    }
 }
 
 /// Similar to dictionary_to_dictionary_values, but overflowing cast is wrapped
@@ -43,6 +46,7 @@ pub fn wrapping_dictionary_to_dictionary_values<K: DictionaryKey>(
 ) -> Result<DictionaryArray<K>> {
     let keys = from.keys();
     let values = from.values();
+    let length = values.len();
 
     let values = cast(
         values.as_ref(),
@@ -52,7 +56,10 @@ pub fn wrapping_dictionary_to_dictionary_values<K: DictionaryKey>(
             partial: false,
         },
     )?;
-    Ok(DictionaryArray::from_data(keys.clone(), values))
+    assert_eq!(values.len(), length); // this is guaranteed by `cast`
+    unsafe {
+        DictionaryArray::try_new_unchecked(from.data_type().clone(), keys.clone(), values.clone())
+    }
 }
 
 /// Casts a [`DictionaryArray`] to a new [`DictionaryArray`] backed by a
@@ -64,18 +71,25 @@ pub fn dictionary_to_dictionary_keys<K1, K2>(
     from: &DictionaryArray<K1>,
 ) -> Result<DictionaryArray<K2>>
 where
-    K1: DictionaryKey,
-    K2: DictionaryKey,
+    K1: DictionaryKey + num_traits::NumCast,
+    K2: DictionaryKey + num_traits::NumCast,
 {
     let keys = from.keys();
     let values = from.values();
+    let is_ordered = from.is_ordered();
 
     let casted_keys = primitive_to_primitive::<K1, K2>(keys, &K2::PRIMITIVE.into());
 
     if casted_keys.null_count() > keys.null_count() {
         Err(Error::Overflow)
     } else {
-        Ok(DictionaryArray::from_data(casted_keys, values.clone()))
+        let data_type = DataType::Dictionary(
+            K2::KEY_TYPE,
+            Box::new(values.data_type().clone()),
+            is_ordered,
+        );
+        // some of the values may not fit in `usize` and thus this needs to be checked
+        DictionaryArray::try_new(data_type, casted_keys, values.clone())
     }
 }
 
@@ -89,17 +103,24 @@ where
 {
     let keys = from.keys();
     let values = from.values();
+    let is_ordered = from.is_ordered();
 
     let casted_keys = primitive_as_primitive::<K1, K2>(keys, &K2::PRIMITIVE.into());
 
     if casted_keys.null_count() > keys.null_count() {
         Err(Error::Overflow)
     } else {
-        Ok(DictionaryArray::from_data(casted_keys, values.clone()))
+        let data_type = DataType::Dictionary(
+            K2::KEY_TYPE,
+            Box::new(values.data_type().clone()),
+            is_ordered,
+        );
+        // some of the values may not fit in `usize` and thus this needs to be checked
+        DictionaryArray::try_new(data_type, casted_keys, values.clone())
     }
 }
 
-pub(super) fn dictionary_cast_dyn<K: DictionaryKey>(
+pub(super) fn dictionary_cast_dyn<K: DictionaryKey + num_traits::NumCast>(
     array: &dyn Array,
     to_type: &DataType,
     options: CastOptions,
@@ -117,6 +138,7 @@ pub(super) fn dictionary_cast_dyn<K: DictionaryKey>(
             match_integer_type!(to_keys_type, |$T| {
                 key_cast!(keys, values, array, &data_type, $T)
             })
+            .map(|x| x.boxed())
         }
         _ => unpack_dictionary::<K>(keys, values.as_ref(), to_type, options),
     }
@@ -130,7 +152,7 @@ fn unpack_dictionary<K>(
     options: CastOptions,
 ) -> Result<Box<dyn Array>>
 where
-    K: DictionaryKey,
+    K: DictionaryKey + num_traits::NumCast,
 {
     // attempt to cast the dict values to the target type
     // use the take kernel to expand out the dictionary
@@ -146,7 +168,7 @@ where
 /// The resulting array has the same length.
 pub fn dictionary_to_values<K>(from: &DictionaryArray<K>) -> Box<dyn Array>
 where
-    K: DictionaryKey,
+    K: DictionaryKey + num_traits::NumCast,
 {
     // take requires first casting i64
     let indices = primitive_to_primitive::<_, i64>(from.keys(), &DataType::Int64);
diff --git a/src/compute/sort/utf8.rs b/src/compute/sort/utf8.rs
index 4d00dc42ec..e2e2da1bc5 100644
--- a/src/compute/sort/utf8.rs
+++ b/src/compute/sort/utf8.rs
@@ -28,10 +28,13 @@ pub(super) fn indices_sorted_unstable_by_dictionary<I: Index, K: DictionaryKey,
         .downcast_ref::<Utf8Array<O>>()
         .unwrap();
 
-    let get = |idx| unsafe {
-        let index = keys.value_unchecked(idx as usize);
-        // Note: there is no check that the keys are within bounds of the dictionary.
-        dict.value(index.to_usize().unwrap())
+    let get = |index| unsafe {
+        // safety: indices_sorted_unstable_by is guaranteed to get items in bounds
+        let index = keys.value_unchecked(index);
+        // safety: dictionaries are guaranteed to have valid usize keys
+        let index = index.as_usize();
+        // safety: dictionaries are guaranteed to have keys in bounds
+        dict.value_unchecked(index)
     };
 
     let cmp = |lhs: &&str, rhs: &&str| lhs.cmp(rhs);
diff --git a/src/compute/take/dict.rs b/src/compute/take/dict.rs
index aa602850f8..1d1efc4fb3 100644
--- a/src/compute/take/dict.rs
+++ b/src/compute/take/dict.rs
@@ -30,5 +30,13 @@ where
     I: Index,
 {
     let keys = take_primitive::<K, I>(values.keys(), indices);
-    DictionaryArray::<K>::from_data(keys, values.values().clone())
+    // safety - this operation takes a subset of keys and thus preserves the dictionary's invariant
+    unsafe {
+        DictionaryArray::<K>::try_new_unchecked(
+            values.data_type().clone(),
+            keys,
+            values.values().clone(),
+        )
+        .unwrap()
+    }
 }
diff --git a/src/io/avro/read/nested.rs b/src/io/avro/read/nested.rs
index e6138e74d3..450930cad7 100644
--- a/src/io/avro/read/nested.rs
+++ b/src/io/avro/read/nested.rs
@@ -157,17 +157,25 @@ impl MutableArray for FixedItemsUtf8Dictionary {
     }
 
     fn as_box(&mut self) -> Box<dyn Array> {
-        Box::new(DictionaryArray::from_data(
-            std::mem::take(&mut self.keys).into(),
-            Box::new(self.values.clone()),
-        ))
+        Box::new(
+            DictionaryArray::try_new(
+                self.data_type.clone(),
+                std::mem::take(&mut self.keys).into(),
+                Box::new(self.values.clone()),
+            )
+            .unwrap(),
+        )
     }
 
     fn as_arc(&mut self) -> std::sync::Arc<dyn Array> {
-        std::sync::Arc::new(DictionaryArray::from_data(
-            std::mem::take(&mut self.keys).into(),
-            Box::new(self.values.clone()),
-        ))
+        std::sync::Arc::new(
+            DictionaryArray::try_new(
+                self.data_type.clone(),
+                std::mem::take(&mut self.keys).into(),
+                Box::new(self.values.clone()),
+            )
+            .unwrap(),
+        )
     }
 
     fn data_type(&self) -> &DataType {
diff --git a/src/io/csv/write/serialize.rs b/src/io/csv/write/serialize.rs
index f1031a979c..6f704d1cc0 100644
--- a/src/io/csv/write/serialize.rs
+++ b/src/io/csv/write/serialize.rs
@@ -471,17 +471,15 @@ fn serialize_utf8_dict<'a, K: DictionaryKey, O: Offset>(
     array: &'a dyn Any,
 ) -> Box<dyn StreamingIterator<Item = [u8]> + 'a> {
     let array = array.downcast_ref::<DictionaryArray<K>>().unwrap();
-    let keys = array.keys();
     let values = array
         .values()
         .as_any()
         .downcast_ref::<Utf8Array<O>>()
         .unwrap();
     Box::new(BufStreamingIterator::new(
-        keys.iter(),
+        array.keys_iter(),
         move |x, buf| {
-            if let Some(x) = x {
-                let i = x.to_usize().unwrap();
+            if let Some(i) = x {
                 if !values.is_null(i) {
                     let val = values.value(i);
                     buf.extend_from_slice(val.as_bytes());
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..f3d92f65a2 100644
--- a/src/io/ipc/read/array/dictionary.rs
+++ b/src/io/ipc/read/array/dictionary.rs
@@ -3,6 +3,7 @@ use std::convert::TryInto;
 use std::io::{Read, Seek};
 
 use crate::array::{DictionaryArray, DictionaryKey};
+use crate::datatypes::DataType;
 use crate::error::{Error, Result};
 
 use super::super::Dictionaries;
@@ -12,12 +13,14 @@ use super::{read_primitive, skip_primitive};
 #[allow(clippy::too_many_arguments)]
 pub fn read_dictionary<T: DictionaryKey, R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
+    data_type: DataType,
     id: Option<i64>,
     buffers: &mut VecDeque<IpcBuffer>,
     reader: &mut R,
     dictionaries: &Dictionaries,
     block_offset: u64,
     compression: Option<Compression>,
+    limit: Option<usize>,
     is_little_endian: bool,
     scratch: &mut Vec<u8>,
 ) -> Result<DictionaryArray<T>>
@@ -48,10 +51,11 @@ where
         block_offset,
         is_little_endian,
         compression,
+        limit,
         scratch,
     )?;
 
-    Ok(DictionaryArray::<T>::from_data(keys, values))
+    DictionaryArray::<T>::try_new(data_type, keys, values)
 }
 
 pub fn skip_dictionary(
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..13300033b6 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,
         )
@@ -180,14 +192,16 @@ pub fn read<R: Read + Seek>(
             match_integer_type!(key_type, |$T| {
                 read_dictionary::<$T, _>(
                     field_nodes,
+                    data_type,
                     ipc_field.dictionary_id,
                     buffers,
                     reader,
                     dictionaries,
                     block_offset,
                     compression,
+                    limit,
                     is_little_endian,
-                    scratch
+                    scratch,
                 )
                 .map(|x| x.boxed())
             })
@@ -202,6 +216,7 @@ pub fn read<R: Read + Seek>(
             block_offset,
             is_little_endian,
             compression,
+            limit,
             version,
             scratch,
         )
@@ -216,6 +231,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/json/read/deserialize.rs b/src/io/json/read/deserialize.rs
index aba6cc6f01..b7add12bc9 100644
--- a/src/io/json/read/deserialize.rs
+++ b/src/io/json/read/deserialize.rs
@@ -282,7 +282,7 @@ fn deserialize_dictionary<'a, K: DictionaryKey, A: Borrow<Value<'a>>>(
     rows: &[A],
     data_type: DataType,
 ) -> DictionaryArray<K> {
-    let child = DictionaryArray::<K>::get_child(&data_type);
+    let child = DictionaryArray::<K>::try_get_child(&data_type).unwrap();
 
     let mut map = HashedMap::<u64, K>::default();
 
@@ -296,8 +296,11 @@ fn deserialize_dictionary<'a, K: DictionaryKey, A: Borrow<Value<'a>>>(
             Some((hash, v)) => match map.get(&hash) {
                 Some(key) => Some(*key),
                 None => {
-                    // todo: convert this to an error.
-                    let key = K::from_usize(map.len()).unwrap();
+                    let key = match map.len().try_into() {
+                        Ok(key) => key,
+                        // todo: convert this to an error.
+                        Err(_) => panic!("The maximum key is too small for this json struct"),
+                    };
                     inner.push(v);
                     map.insert(hash, key);
                     Some(key)
@@ -307,8 +310,9 @@ fn deserialize_dictionary<'a, K: DictionaryKey, A: Borrow<Value<'a>>>(
         })
         .collect::<PrimitiveArray<K>>();
 
+    drop(extractor);
     let values = _deserialize(&inner, child.clone());
-    DictionaryArray::<K>::from_data(keys, values)
+    DictionaryArray::<K>::try_new(data_type, keys, values).unwrap()
 }
 
 pub(crate) fn _deserialize<'a, A: Borrow<Value<'a>>>(
diff --git a/src/io/json_integration/read/array.rs b/src/io/json_integration/read/array.rs
index e434cbe691..2d7d0ec39a 100644
--- a/src/io/json_integration/read/array.rs
+++ b/src/io/json_integration/read/array.rs
@@ -230,7 +230,7 @@ fn to_map(
     Ok(Box::new(MapArray::new(data_type, offsets, field, validity)))
 }
 
-fn to_dictionary<K: DictionaryKey>(
+fn to_dictionary<K: DictionaryKey + NumCast>(
     data_type: DataType,
     field: &IpcField,
     json_col: &ArrowJsonColumn,
@@ -244,7 +244,7 @@ fn to_dictionary<K: DictionaryKey>(
 
     let keys = to_primitive(json_col, K::PRIMITIVE.into());
 
-    let inner_data_type = DictionaryArray::<K>::get_child(&data_type);
+    let inner_data_type = DictionaryArray::<K>::try_get_child(&data_type)?;
     let values = to_array(
         inner_data_type.clone(),
         field,
@@ -252,7 +252,7 @@ fn to_dictionary<K: DictionaryKey>(
         dictionaries,
     )?;
 
-    Ok(Box::new(DictionaryArray::<K>::from_data(keys, values)))
+    DictionaryArray::<K>::try_new(data_type, keys, values).map(|a| a.boxed())
 }
 
 /// Construct an [`Array`] from the JSON integration format
diff --git a/src/io/parquet/read/deserialize/binary/dictionary.rs b/src/io/parquet/read/deserialize/binary/dictionary.rs
index b5448a077b..bf656929e6 100644
--- a/src/io/parquet/read/deserialize/binary/dictionary.rs
+++ b/src/io/parquet/read/deserialize/binary/dictionary.rs
@@ -23,6 +23,7 @@ where
 {
     iter: I,
     data_type: DataType,
+    values_data_type: DataType,
     values: Dict,
     items: VecDeque<(Vec<K>, MutableBitmap)>,
     chunk_size: Option<usize>,
@@ -36,13 +37,14 @@ where
     I: DataPages,
 {
     pub fn new(iter: I, data_type: DataType, chunk_size: Option<usize>) -> Self {
-        let data_type = match data_type {
+        let values_data_type = match &data_type {
             DataType::Dictionary(_, values, _) => values.as_ref().clone(),
             _ => unreachable!(),
         };
         Self {
             iter,
             data_type,
+            values_data_type,
             values: Dict::Empty,
             items: VecDeque::new(),
             chunk_size,
@@ -90,8 +92,9 @@ where
             &mut self.iter,
             &mut self.items,
             &mut self.values,
+            self.data_type.clone(),
             self.chunk_size,
-            |dict| read_dict::<O>(self.data_type.clone(), dict),
+            |dict| read_dict::<O>(self.values_data_type.clone(), dict),
         );
         match maybe_state {
             MaybeNext::Some(Ok(dict)) => Some(Ok(dict)),
diff --git a/src/io/parquet/read/deserialize/dictionary.rs b/src/io/parquet/read/deserialize/dictionary.rs
index 62ef51adc9..f89a5e06d5 100644
--- a/src/io/parquet/read/deserialize/dictionary.rs
+++ b/src/io/parquet/read/deserialize/dictionary.rs
@@ -10,6 +10,7 @@ use parquet2::{
 use crate::{
     array::{Array, DictionaryArray, DictionaryKey, PrimitiveArray},
     bitmap::MutableBitmap,
+    datatypes::DataType,
     error::{Error, Result},
 };
 
@@ -158,13 +159,30 @@ where
                 &mut page.validity,
                 Some(remaining),
                 values,
-                &mut page.values.by_ref().map(|x| K::from_u32(x).unwrap()),
+                &mut page.values.by_ref().map(|x| {
+                    let x: usize = x.try_into().unwrap();
+                    match x.try_into() {
+                        Ok(key) => key,
+                        // todo: convert this to an error.
+                        Err(_) => panic!("The maximum key is too small"),
+                    }
+                }),
             ),
             State::Required(page) => {
                 values.extend(
                     page.values
                         .by_ref()
-                        .map(|x| K::from_u32(x).unwrap())
+                        .map(|x| {
+                            let x: usize = x.try_into().unwrap();
+                            let x: K = match x.try_into() {
+                                Ok(key) => key,
+                                // todo: convert this to an error.
+                                Err(_) => {
+                                    panic!("The maximum key is too small")
+                                }
+                            };
+                            x
+                        })
                         .take(remaining),
                 );
             }
@@ -173,13 +191,33 @@ where
                 page_validity,
                 Some(remaining),
                 values,
-                &mut page_values.by_ref().map(|x| K::from_u32(x).unwrap()),
+                &mut page_values.by_ref().map(|x| {
+                    let x: usize = x.try_into().unwrap();
+                    let x: K = match x.try_into() {
+                        Ok(key) => key,
+                        // todo: convert this to an error.
+                        Err(_) => {
+                            panic!("The maximum key is too small")
+                        }
+                    };
+                    x
+                }),
             ),
             State::FilteredRequired(page) => {
                 values.extend(
                     page.values
                         .by_ref()
-                        .map(|x| K::from_u32(x).unwrap())
+                        .map(|x| {
+                            let x: usize = x.try_into().unwrap();
+                            let x: K = match x.try_into() {
+                                Ok(key) => key,
+                                // todo: convert this to an error.
+                                Err(_) => {
+                                    panic!("The maximum key is too small")
+                                }
+                            };
+                            x
+                        })
                         .take(remaining),
                 );
             }
@@ -203,7 +241,7 @@ impl Dict {
 }
 
 fn finish_key<K: DictionaryKey>(values: Vec<K>, validity: MutableBitmap) -> PrimitiveArray<K> {
-    PrimitiveArray::from_data(K::PRIMITIVE.into(), values.into(), validity.into())
+    PrimitiveArray::new(K::PRIMITIVE.into(), values.into(), validity.into())
 }
 
 #[inline]
@@ -216,13 +254,14 @@ pub(super) fn next_dict<
     iter: &'a mut I,
     items: &mut VecDeque<(Vec<K>, MutableBitmap)>,
     dict: &mut Dict,
+    data_type: DataType,
     chunk_size: Option<usize>,
     read_dict: F,
 ) -> MaybeNext<Result<DictionaryArray<K>>> {
     if items.len() > 1 {
         let (values, validity) = items.pop_front().unwrap();
         let keys = finish_key(values, validity);
-        return MaybeNext::Some(Ok(DictionaryArray::from_data(keys, dict.unwrap())));
+        return MaybeNext::Some(DictionaryArray::try_new(data_type, keys, dict.unwrap()));
     }
     match iter.next() {
         Err(e) => MaybeNext::Some(Err(e.into())),
@@ -255,7 +294,7 @@ pub(super) fn next_dict<
                 let (values, validity) = items.pop_front().unwrap();
                 let keys =
                     PrimitiveArray::from_data(K::PRIMITIVE.into(), values.into(), validity.into());
-                MaybeNext::Some(Ok(DictionaryArray::from_data(keys, dict.unwrap())))
+                MaybeNext::Some(DictionaryArray::try_new(data_type, keys, dict.unwrap()))
             }
         }
         Ok(None) => {
@@ -266,7 +305,7 @@ pub(super) fn next_dict<
 
                 let keys = finish_key(values, validity);
 
-                MaybeNext::Some(Ok(DictionaryArray::from_data(keys, dict.unwrap())))
+                MaybeNext::Some(DictionaryArray::try_new(data_type, keys, dict.unwrap()))
             } else {
                 MaybeNext::None
             }
diff --git a/src/io/parquet/read/deserialize/fixed_size_binary/dictionary.rs b/src/io/parquet/read/deserialize/fixed_size_binary/dictionary.rs
index 1db6b48003..4d44ef4f72 100644
--- a/src/io/parquet/read/deserialize/fixed_size_binary/dictionary.rs
+++ b/src/io/parquet/read/deserialize/fixed_size_binary/dictionary.rs
@@ -22,6 +22,7 @@ where
 {
     iter: I,
     data_type: DataType,
+    values_data_type: DataType,
     values: Dict,
     items: VecDeque<(Vec<K>, MutableBitmap)>,
     chunk_size: Option<usize>,
@@ -33,13 +34,14 @@ where
     I: DataPages,
 {
     pub fn new(iter: I, data_type: DataType, chunk_size: Option<usize>) -> Self {
-        let data_type = match data_type {
+        let values_data_type = match &data_type {
             DataType::Dictionary(_, values, _) => values.as_ref().clone(),
             _ => unreachable!(),
         };
         Self {
             iter,
             data_type,
+            values_data_type,
             values: Dict::Empty,
             items: VecDeque::new(),
             chunk_size,
@@ -73,8 +75,9 @@ where
             &mut self.iter,
             &mut self.items,
             &mut self.values,
+            self.data_type.clone(),
             self.chunk_size,
-            |dict| read_dict(self.data_type.clone(), dict),
+            |dict| read_dict(self.values_data_type.clone(), dict),
         );
         match maybe_state {
             MaybeNext::Some(Ok(dict)) => Some(Ok(dict)),
diff --git a/src/io/parquet/read/deserialize/primitive/dictionary.rs b/src/io/parquet/read/deserialize/primitive/dictionary.rs
index c729363379..2b3f0ca549 100644
--- a/src/io/parquet/read/deserialize/primitive/dictionary.rs
+++ b/src/io/parquet/read/deserialize/primitive/dictionary.rs
@@ -45,6 +45,7 @@ where
 {
     iter: I,
     data_type: DataType,
+    values_data_type: DataType,
     values: Dict,
     items: VecDeque<(Vec<K>, MutableBitmap)>,
     chunk_size: Option<usize>,
@@ -62,13 +63,14 @@ where
     F: Copy + Fn(P) -> T,
 {
     pub fn new(iter: I, data_type: DataType, chunk_size: Option<usize>, op: F) -> Self {
-        let data_type = match data_type {
-            DataType::Dictionary(_, values, _) => *values,
-            _ => data_type,
+        let values_data_type = match &data_type {
+            DataType::Dictionary(_, values, _) => *(values.clone()),
+            _ => unreachable!(),
         };
         Self {
             iter,
             data_type,
+            values_data_type,
             values: Dict::Empty,
             items: VecDeque::new(),
             chunk_size,
@@ -93,8 +95,9 @@ where
             &mut self.iter,
             &mut self.items,
             &mut self.values,
+            self.data_type.clone(),
             self.chunk_size,
-            |dict| read_dict::<P, T, _>(self.data_type.clone(), self.op, dict),
+            |dict| read_dict::<P, T, _>(self.values_data_type.clone(), self.op, dict),
         );
         match maybe_state {
             MaybeNext::Some(Ok(dict)) => Some(Ok(dict)),
diff --git a/src/io/parquet/read/statistics/dictionary.rs b/src/io/parquet/read/statistics/dictionary.rs
index d29feb4680..b2553c62f1 100644
--- a/src/io/parquet/read/statistics/dictionary.rs
+++ b/src/io/parquet/read/statistics/dictionary.rs
@@ -41,7 +41,7 @@ impl MutableArray for DynMutableDictionary {
         match self.data_type.to_physical_type() {
             PhysicalType::Dictionary(key) => match_integer_type!(key, |$T| {
                 let keys = PrimitiveArray::<$T>::from_iter((0..inner.len() as $T).map(Some));
-                Box::new(DictionaryArray::<$T>::from_data(keys, inner))
+                Box::new(DictionaryArray::<$T>::try_new(self.data_type.clone(), keys, inner).unwrap())
             }),
             _ => todo!(),
         }
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(_, _, _) => {
diff --git a/src/io/parquet/write/dictionary.rs b/src/io/parquet/write/dictionary.rs
index 05c56207bf..f6a9bcabbb 100644
--- a/src/io/parquet/write/dictionary.rs
+++ b/src/io/parquet/write/dictionary.rs
@@ -20,26 +20,25 @@ use crate::datatypes::DataType;
 use crate::error::{Error, Result};
 use crate::io::parquet::write::utils;
 use crate::{
-    array::{Array, DictionaryArray, DictionaryKey, PrimitiveArray},
+    array::{Array, DictionaryArray, DictionaryKey},
     io::parquet::read::schema::is_nullable,
 };
 
 fn encode_keys<K: DictionaryKey>(
-    array: &PrimitiveArray<K>,
-    validity: Option<&Bitmap>,
+    array: &DictionaryArray<K>,
     type_: PrimitiveType,
     statistics: ParquetStatistics,
     options: WriteOptions,
 ) -> Result<EncodedPage> {
+    let validity = array.values().validity();
     let is_optional = is_nullable(&type_.field_info);
 
     let mut buffer = vec![];
 
     let null_count = if let Some(validity) = validity {
-        let projected_validity = array.iter().map(|x| {
-            x.map(|x| validity.get_bit(x.to_usize().unwrap()))
-                .unwrap_or(false)
-        });
+        let projected_validity = array
+            .keys_iter()
+            .map(|x| x.map(|x| validity.get_bit(x)).unwrap_or(false));
         let projected_val = Bitmap::from_trusted_len_iter(projected_validity);
 
         let null_count = projected_val.unset_bits();
@@ -68,8 +67,7 @@ fn encode_keys<K: DictionaryKey>(
     // encode indices
     // compute the required number of bits
     if let Some(validity) = validity {
-        let keys = array.iter().flatten().filter_map(|x| {
-            let index = x.to_usize().unwrap();
+        let keys = array.keys_iter().flatten().filter_map(|index| {
             // discard indices whose values are null, since they are part of the def levels.
             if validity.get_bit(index) {
                 Some(index as u32)
@@ -87,7 +85,7 @@ fn encode_keys<K: DictionaryKey>(
         // followed by the encoded indices.
         encode_u32(&mut buffer, keys, num_bits)?;
     } else {
-        let keys = array.iter().flatten().map(|x| x.to_usize().unwrap() as u32);
+        let keys = array.keys_iter().flatten().map(|x| x as u32);
         let num_bits = utils::get_bit_width(keys.clone().max().unwrap_or(0) as u64) as u8;
 
         let keys = utils::ExactSizedIter::new(keys, array.len() - array.null_count());
@@ -202,13 +200,7 @@ pub fn array_to_pages<K: DictionaryKey>(
             let dict_page = EncodedPage::Dict(dict_page);
 
             // write DataPage pointing to DictPage
-            let data_page = encode_keys(
-                array.keys(),
-                array.values().validity(),
-                type_,
-                statistics,
-                options,
-            )?;
+            let data_page = encode_keys(array, type_, statistics, options)?;
 
             let iter = std::iter::once(Ok(dict_page)).chain(std::iter::once(Ok(data_page)));
             Ok(DynIter::new(Box::new(iter)))


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/array/dictionary/mod.rs b/tests/it/array/dictionary/mod.rs
index f7a9b61988..f436e2685d 100644
--- a/tests/it/array/dictionary/mod.rs
+++ b/tests/it/array/dictionary/mod.rs
@@ -1,1 +1,153 @@
 mod mutable;
+
+use arrow2::{array::*, datatypes::DataType};
+
+#[test]
+fn try_new_ok() {
+    let values = Utf8Array::<i32>::from_slice(&["a", "aa"]);
+    let data_type =
+        DataType::Dictionary(i32::KEY_TYPE, Box::new(values.data_type().clone()), false);
+    let array = DictionaryArray::try_new(
+        data_type,
+        PrimitiveArray::from_vec(vec![1, 0]),
+        values.boxed(),
+    )
+    .unwrap();
+
+    assert_eq!(array.keys(), &PrimitiveArray::from_vec(vec![1i32, 0]));
+    assert_eq!(
+        &Utf8Array::<i32>::from_slice(&["a", "aa"]) as &dyn Array,
+        array.values().as_ref(),
+    );
+    assert!(!array.is_ordered());
+
+    assert_eq!(format!("{:?}", array), "DictionaryArray[aa, a]");
+}
+
+#[test]
+fn try_new_incorrect_key() {
+    let values = Utf8Array::<i32>::from_slice(&["a", "aa"]);
+    let data_type =
+        DataType::Dictionary(i16::KEY_TYPE, Box::new(values.data_type().clone()), false);
+
+    let r = DictionaryArray::try_new(
+        data_type,
+        PrimitiveArray::from_vec(vec![1, 0]),
+        values.boxed(),
+    )
+    .is_err();
+
+    assert!(r);
+}
+
+#[test]
+fn try_new_incorrect_dt() {
+    let values = Utf8Array::<i32>::from_slice(&["a", "aa"]);
+    let data_type = DataType::Int32;
+
+    let r = DictionaryArray::try_new(
+        data_type,
+        PrimitiveArray::from_vec(vec![1, 0]),
+        values.boxed(),
+    )
+    .is_err();
+
+    assert!(r);
+}
+
+#[test]
+fn try_new_incorrect_values_dt() {
+    let values = Utf8Array::<i32>::from_slice(&["a", "aa"]);
+    let data_type = DataType::Dictionary(i32::KEY_TYPE, Box::new(DataType::LargeUtf8), false);
+
+    let r = DictionaryArray::try_new(
+        data_type,
+        PrimitiveArray::from_vec(vec![1, 0]),
+        values.boxed(),
+    )
+    .is_err();
+
+    assert!(r);
+}
+
+#[test]
+fn try_new_out_of_bounds() {
+    let values = Utf8Array::<i32>::from_slice(&["a", "aa"]);
+
+    let r = DictionaryArray::try_from_keys(PrimitiveArray::from_vec(vec![2, 0]), values.boxed())
+        .is_err();
+
+    assert!(r);
+}
+
+#[test]
+fn try_new_out_of_bounds_neg() {
+    let values = Utf8Array::<i32>::from_slice(&["a", "aa"]);
+
+    let r = DictionaryArray::try_from_keys(PrimitiveArray::from_vec(vec![-1, 0]), values.boxed())
+        .is_err();
+
+    assert!(r);
+}
+
+#[test]
+fn new_null() {
+    let dt = DataType::Dictionary(i16::KEY_TYPE, Box::new(DataType::Int32), false);
+    let array = DictionaryArray::<i16>::new_null(dt, 2);
+
+    assert_eq!(format!("{:?}", array), "DictionaryArray[None, None]");
+}
+
+#[test]
+fn new_empty() {
+    let dt = DataType::Dictionary(i16::KEY_TYPE, Box::new(DataType::Int32), false);
+    let array = DictionaryArray::<i16>::new_empty(dt);
+
+    assert_eq!(format!("{:?}", array), "DictionaryArray[]");
+}
+
+#[test]
+fn with_validity() {
+    let values = Utf8Array::<i32>::from_slice(&["a", "aa"]);
+    let array =
+        DictionaryArray::try_from_keys(PrimitiveArray::from_vec(vec![1, 0]), values.boxed())
+            .unwrap();
+
+    let array = array.with_validity(Some([true, false].into()));
+
+    assert_eq!(format!("{:?}", array), "DictionaryArray[aa, None]");
+}
+
+#[test]
+fn rev_iter() {
+    let values = Utf8Array::<i32>::from_slice(&["a", "aa"]);
+    let array =
+        DictionaryArray::try_from_keys(PrimitiveArray::from_vec(vec![1, 0]), values.boxed())
+            .unwrap();
+
+    let mut iter = array.into_iter();
+    assert_eq!(iter.by_ref().rev().count(), 2);
+    assert_eq!(iter.size_hint(), (0, Some(0)));
+}
+
+#[test]
+fn iter_values() {
+    let values = Utf8Array::<i32>::from_slice(&["a", "aa"]);
+    let array =
+        DictionaryArray::try_from_keys(PrimitiveArray::from_vec(vec![1, 0]), values.boxed())
+            .unwrap();
+
+    let mut iter = array.values_iter();
+    assert_eq!(iter.by_ref().count(), 2);
+    assert_eq!(iter.size_hint(), (0, Some(0)));
+}
+
+#[test]
+fn keys_values_iter() {
+    let values = Utf8Array::<i32>::from_slice(&["a", "aa"]);
+    let array =
+        DictionaryArray::try_from_keys(PrimitiveArray::from_vec(vec![1, 0]), values.boxed())
+            .unwrap();
+
+    assert_eq!(array.keys_values_iter().collect::<Vec<_>>(), vec![1, 0]);
+}
diff --git a/tests/it/array/equal/dictionary.rs b/tests/it/array/equal/dictionary.rs
index 1bbe9627d6..8e25d083e1 100644
--- a/tests/it/array/equal/dictionary.rs
+++ b/tests/it/array/equal/dictionary.rs
@@ -6,7 +6,7 @@ fn create_dictionary_array(values: &[Option<&str>], keys: &[Option<i16>]) -> Dic
     let keys = Int16Array::from(keys);
     let values = Utf8Array::<i32>::from(values);
 
-    DictionaryArray::from_data(keys, Box::new(values))
+    DictionaryArray::try_from_keys(keys, values.boxed()).unwrap()
 }
 
 #[test]
diff --git a/tests/it/array/growable/dictionary.rs b/tests/it/array/growable/dictionary.rs
index a45c0b298b..5d0eeef226 100644
--- a/tests/it/array/growable/dictionary.rs
+++ b/tests/it/array/growable/dictionary.rs
@@ -1,6 +1,5 @@
 use arrow2::array::growable::{Growable, GrowableDictionary};
 use arrow2::array::*;
-use arrow2::datatypes::DataType;
 use arrow2::error::Result;
 
 #[test]
@@ -13,10 +12,11 @@ fn test_single() -> Result<()> {
     let array = array.into();
 
     // same values, less keys
-    let expected = DictionaryArray::<i32>::from_data(
-        PrimitiveArray::from(vec![Some(1), Some(0)]),
+    let expected = DictionaryArray::try_from_keys(
+        PrimitiveArray::from_vec(vec![1, 0]),
         Box::new(Utf8Array::<i32>::from(&original_data)),
-    );
+    )
+    .unwrap();
 
     let mut growable = GrowableDictionary::new(&[&array], false, 0);
 
@@ -28,25 +28,6 @@ fn test_single() -> Result<()> {
     Ok(())
 }
 
-#[test]
-fn test_negative_keys() {
-    let vals = vec![Some("a"), Some("b"), Some("c")];
-    let keys = vec![0, 1, 2, -1];
-
-    let keys = PrimitiveArray::from_data(
-        DataType::Int32,
-        keys.into(),
-        Some(vec![true, true, true, false].into()),
-    );
-
-    let arr = DictionaryArray::from_data(keys, Box::new(Utf8Array::<i32>::from(vals)));
-    // check that we don't panic with negative keys to usize conversion
-    let mut growable = GrowableDictionary::new(&[&arr], false, 0);
-    growable.extend(0, 0, 4);
-    let out: DictionaryArray<i32> = growable.into();
-    assert_eq!(out, arr);
-}
-
 #[test]
 fn test_multi() -> Result<()> {
     let mut original_data1 = vec![Some("a"), Some("b"), None, Some("a")];
@@ -65,10 +46,11 @@ fn test_multi() -> Result<()> {
 
     // same values, less keys
     original_data1.extend(original_data2.iter().cloned());
-    let expected = DictionaryArray::<i32>::from_data(
-        PrimitiveArray::from(vec![Some(1), None, Some(3), None]),
-        Box::new(Utf8Array::<i32>::from_slice(&["a", "b", "c", "b", "a"])),
-    );
+    let expected = DictionaryArray::try_from_keys(
+        PrimitiveArray::from(&[Some(1), None, Some(3), None]),
+        Utf8Array::<i32>::from_slice(&["a", "b", "c", "b", "a"]).boxed(),
+    )
+    .unwrap();
 
     let mut growable = GrowableDictionary::new(&[&array1, &array2], false, 0);
 
diff --git a/tests/it/array/growable/mod.rs b/tests/it/array/growable/mod.rs
index acc5da6ddc..a6e25ecba3 100644
--- a/tests/it/array/growable/mod.rs
+++ b/tests/it/array/growable/mod.rs
@@ -38,9 +38,10 @@ fn test_make_growable() {
         FixedSizeBinaryArray::new(DataType::FixedSizeBinary(2), b"abcd".to_vec().into(), None);
     make_growable(&[&array], false, 2);
 
-    let array = DictionaryArray::<i32>::from_data(
-        Int32Array::from_slice([1, 2]),
-        Box::new(Int32Array::from_slice([1, 2])),
-    );
+    let array = DictionaryArray::try_from_keys(
+        Int32Array::from_slice([1, 0]),
+        Int32Array::from_slice([1, 2]).boxed(),
+    )
+    .unwrap();
     make_growable(&[&array], false, 2);
 }
diff --git a/tests/it/compute/arithmetics/mod.rs b/tests/it/compute/arithmetics/mod.rs
index f697cd8041..a22f146d7b 100644
--- a/tests/it/compute/arithmetics/mod.rs
+++ b/tests/it/compute/arithmetics/mod.rs
@@ -95,14 +95,16 @@ fn test_neg() {
 
 #[test]
 fn test_neg_dict() {
-    let a = DictionaryArray::<u8>::from_data(
+    let a = DictionaryArray::try_from_keys(
         UInt8Array::from_slice(&[0, 0, 1]),
-        Box::new(Int8Array::from_slice(&[1, 2])),
-    );
+        Int8Array::from_slice(&[1, 2]).boxed(),
+    )
+    .unwrap();
     let result = neg(&a);
-    let expected = DictionaryArray::<u8>::from_data(
+    let expected = DictionaryArray::try_from_keys(
         UInt8Array::from_slice(&[0, 0, 1]),
-        Box::new(Int8Array::from_slice(&[-1, -2])),
-    );
+        Int8Array::from_slice(&[-1, -2]).boxed(),
+    )
+    .unwrap();
     assert_eq!(expected, result.as_ref());
 }
diff --git a/tests/it/io/avro/read.rs b/tests/it/io/avro/read.rs
index bef9dcb252..deb8ffd802 100644
--- a/tests/it/io/avro/read.rs
+++ b/tests/it/io/avro/read.rs
@@ -106,10 +106,11 @@ pub(super) fn data() -> Chunk<Box<dyn Array>> {
             None,
         )
         .boxed(),
-        DictionaryArray::<i32>::from_data(
+        DictionaryArray::try_from_keys(
             Int32Array::from_slice([1, 0]),
             Box::new(Utf8Array::<i32>::from_slice(["SPADES", "HEARTS"])),
         )
+        .unwrap()
         .boxed(),
         PrimitiveArray::<i128>::from_slice([12345678i128, -12345678i128])
             .to(DataType::Decimal(18, 5))
diff --git a/tests/it/io/csv/write.rs b/tests/it/io/csv/write.rs
index 44bec50497..80d7212175 100644
--- a/tests/it/io/csv/write.rs
+++ b/tests/it/io/csv/write.rs
@@ -16,7 +16,7 @@ fn data() -> Chunk<Box<dyn Array>> {
     let c6 = PrimitiveArray::<i32>::from_vec(vec![1234, 24680, 85563])
         .to(DataType::Time32(TimeUnit::Second));
     let keys = UInt32Array::from_slice(&[2, 0, 1]);
-    let c7 = DictionaryArray::from_data(keys, Box::new(c1.clone()));
+    let c7 = DictionaryArray::try_from_keys(keys, Box::new(c1.clone())).unwrap();
 
     Chunk::new(vec![
         Box::new(c1) as Box<dyn Array>,
@@ -256,13 +256,13 @@ fn data_array(column: &str) -> (Chunk<Box<dyn Array>>, Vec<&'static str>) {
         "dictionary[u32]" => {
             let keys = UInt32Array::from_slice(&[2, 1, 0]);
             let values = Utf8Array::<i64>::from_slice(["a b", "c", "d"]).boxed();
-            let array = DictionaryArray::from_data(keys, values);
+            let array = DictionaryArray::try_from_keys(keys, values).unwrap();
             (array.boxed(), vec!["d", "c", "a b"])
         }
         "dictionary[u64]" => {
             let keys = UInt64Array::from_slice(&[2, 1, 0]);
             let values = Utf8Array::<i64>::from_slice(["a b", "c", "d"]).boxed();
-            let array = DictionaryArray::from_data(keys, values);
+            let array = DictionaryArray::try_from_keys(keys, values).unwrap();
             (array.boxed(), vec!["d", "c", "a b"])
         }
         _ => todo!(),
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..4f3ddd2b6a 100644
--- a/tests/it/io/parquet/mod.rs
+++ b/tests/it/io/parquet/mod.rs
@@ -359,7 +359,7 @@ pub fn pyarrow_nullable(column: &str) -> Box<dyn Array> {
         "int32_dict" => {
             let keys = PrimitiveArray::<i32>::from([Some(0), Some(1), None, Some(1)]);
             let values = Box::new(PrimitiveArray::<i32>::from_slice([10, 200]));
-            Box::new(DictionaryArray::<i32>::from_data(keys, values))
+            Box::new(DictionaryArray::try_from_keys(keys, values).unwrap())
         }
         "decimal_9" => {
             let values = i64_values
@@ -440,10 +440,7 @@ pub fn pyarrow_nullable_statistics(column: &str) -> Statistics {
         },
         "int32_dict" => {
             let new_dict = |array: Box<dyn Array>| -> Box<dyn Array> {
-                Box::new(DictionaryArray::<i32>::from_data(
-                    vec![Some(0)].into(),
-                    array,
-                ))
+                Box::new(DictionaryArray::try_from_keys(vec![Some(0)].into(), array).unwrap())
             };
 
             Statistics {
@@ -563,7 +560,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 +569,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 +663,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 +672,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!(),
     }
@@ -975,42 +996,45 @@ fn arrow_type() -> Result<()> {
     let array2 = Utf8Array::<i64>::from([Some("a"), None, Some("bb")]);
 
     let indices = PrimitiveArray::from_values((0..3u64).map(|x| x % 2));
-    let values = PrimitiveArray::from_slice([1.0f32, 3.0]);
-    let array3 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+    let values = PrimitiveArray::from_slice([1.0f32, 3.0]).boxed();
+    let array3 = DictionaryArray::try_from_keys(indices.clone(), values).unwrap();
 
-    let values = BinaryArray::<i32>::from_slice([b"ab", b"ac"]);
-    let array4 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+    let values = BinaryArray::<i32>::from_slice([b"ab", b"ac"]).boxed();
+    let array4 = DictionaryArray::try_from_keys(indices.clone(), values).unwrap();
 
     let values = FixedSizeBinaryArray::from_data(
         DataType::FixedSizeBinary(2),
         vec![b'a', b'b', b'a', b'c'].into(),
         None,
-    );
-    let array5 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+    )
+    .boxed();
+    let array5 = DictionaryArray::try_from_keys(indices.clone(), values).unwrap();
 
-    let values = PrimitiveArray::from_slice([1i16, 3]);
-    let array6 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+    let values = PrimitiveArray::from_slice([1i16, 3]).boxed();
+    let array6 = DictionaryArray::try_from_keys(indices.clone(), values).unwrap();
 
-    let values = PrimitiveArray::from_slice([1i64, 3]).to(DataType::Timestamp(
-        TimeUnit::Millisecond,
-        Some("UTC".to_string()),
-    ));
-    let array7 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+    let values = PrimitiveArray::from_slice([1i64, 3])
+        .to(DataType::Timestamp(
+            TimeUnit::Millisecond,
+            Some("UTC".to_string()),
+        ))
+        .boxed();
+    let array7 = DictionaryArray::try_from_keys(indices.clone(), values).unwrap();
 
-    let values = PrimitiveArray::from_slice([1.0f64, 3.0]);
-    let array8 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+    let values = PrimitiveArray::from_slice([1.0f64, 3.0]).boxed();
+    let array8 = DictionaryArray::try_from_keys(indices.clone(), values).unwrap();
 
-    let values = PrimitiveArray::from_slice([1u8, 3]);
-    let array9 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+    let values = PrimitiveArray::from_slice([1u8, 3]).boxed();
+    let array9 = DictionaryArray::try_from_keys(indices.clone(), values).unwrap();
 
-    let values = PrimitiveArray::from_slice([1u16, 3]);
-    let array10 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+    let values = PrimitiveArray::from_slice([1u16, 3]).boxed();
+    let array10 = DictionaryArray::try_from_keys(indices.clone(), values).unwrap();
 
-    let values = PrimitiveArray::from_slice([1u32, 3]);
-    let array11 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+    let values = PrimitiveArray::from_slice([1u32, 3]).boxed();
+    let array11 = DictionaryArray::try_from_keys(indices.clone(), values).unwrap();
 
-    let values = PrimitiveArray::from_slice([1u64, 3]);
-    let array12 = DictionaryArray::from_data(indices, Box::new(values));
+    let values = PrimitiveArray::from_slice([1u64, 3]).boxed();
+    let array12 = DictionaryArray::try_from_keys(indices, values).unwrap();
 
     let array13 = PrimitiveArray::<i32>::from_slice([1, 2, 3])
         .to(DataType::Interval(IntervalUnit::YearMonth));
diff --git a/tests/it/io/parquet/read_indexes.rs b/tests/it/io/parquet/read_indexes.rs
index 92b647acf1..ee91ba846a 100644
--- a/tests/it/io/parquet/read_indexes.rs
+++ b/tests/it/io/parquet/read_indexes.rs
@@ -208,12 +208,12 @@ fn indexed_optional_boolean() -> Result<()> {
 #[test]
 fn indexed_dict() -> Result<()> {
     let indices = PrimitiveArray::from_values((0..6u64).map(|x| x % 2));
-    let values = PrimitiveArray::from_slice([4i32, 6i32]);
-    let array = DictionaryArray::from_data(indices, Box::new(values));
+    let values = PrimitiveArray::from_slice([4i32, 6i32]).boxed();
+    let array = DictionaryArray::try_from_keys(indices, values).unwrap();
 
     let indices = PrimitiveArray::from_slice(&[0u64]);
-    let values = PrimitiveArray::from_slice([4i32, 6i32]);
-    let expected = DictionaryArray::from_data(indices, Box::new(values));
+    let values = PrimitiveArray::from_slice([4i32, 6i32]).boxed();
+    let expected = DictionaryArray::try_from_keys(indices, values).unwrap();
 
     let expected = expected.boxed();
 
diff --git a/tests/it/io/print.rs b/tests/it/io/print.rs
index a2f4eb5c9b..6e37836d6b 100644
--- a/tests/it/io/print.rs
+++ b/tests/it/io/print.rs
@@ -97,7 +97,7 @@ fn write_dictionary() -> Result<()> {
 fn dictionary_validities() -> Result<()> {
     let keys = PrimitiveArray::<i32>::from([Some(1), None, Some(0)]);
     let values = PrimitiveArray::<i32>::from([None, Some(10)]);
-    let array = DictionaryArray::<i32>::from_data(keys, Box::new(values));
+    let array = DictionaryArray::try_from_keys(keys, Box::new(values)).unwrap();
 
     let columns = Chunk::new(vec![&array as &dyn Array]);