Repository: jorgecarleitao/arrow2
Pull request #1046 (2022-06-03 15:06:25)

Fixed using lower limit than size of first parquet row group
If the first row-group of parquet data had more rows than the limit, no data would be returned even if the chunk size was available in that limit and first row group


Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 09fae9293a..7b7f9c48f7 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -39,9 +39,7 @@ regex = { version = "^1.3", optional = true }
 streaming-iterator = { version = "0.1", optional = true }
 fallible-streaming-iterator = { version = "0.1", optional = true }
 
-serde = { version = "^1.0", features = ["rc"], optional = true }
-serde_derive = { version = "^1.0", optional = true }
-serde_json = { version = "^1.0", features = ["preserve_order"], optional = true }
+json-deserializer = { version = "0.3", optional = true }
 indexmap = { version = "^1.6", optional = true }
 
 # used to print columns in a nice columnar format
@@ -72,6 +70,9 @@ parquet2 = { version = "0.13", optional = true, default_features = false }
 
 # avro support
 avro-schema = { version = "0.2", optional = true }
+serde = { version = "^1.0", features = ["rc"], optional = true }
+serde_derive = { version = "^1.0", optional = true }
+serde_json = { version = "^1.0", features = ["preserve_order"], optional = true }
 # compression of avro
 libflate = { version = "1.1.1", optional = true }
 snap = { version = "1", optional = true }
@@ -134,7 +135,7 @@ io_csv_async = ["io_csv_read_async"]
 io_csv_read = ["csv", "lexical-core"]
 io_csv_read_async = ["csv-async", "lexical-core", "futures"]
 io_csv_write = ["csv-core", "streaming-iterator", "lexical-core"]
-io_json = ["serde", "serde_json", "streaming-iterator", "fallible-streaming-iterator", "indexmap", "lexical-core"]
+io_json = ["json-deserializer", "streaming-iterator", "fallible-streaming-iterator", "indexmap", "lexical-core"]
 io_ipc = ["arrow-format"]
 io_ipc_write_async = ["io_ipc", "futures"]
 io_ipc_read_async = ["io_ipc", "futures", "async-stream"]
@@ -156,9 +157,9 @@ io_avro_compression = [
     "crc",
 ]
 io_avro_async = ["io_avro", "futures", "async-stream"]
-# io_json: its dependencies + error handling
+# serde+serde_json: its dependencies + error handling
 # serde_derive: there is some derive around
-io_json_integration = ["io_json", "serde_derive", "hex"]
+io_json_integration = ["hex", "serde", "serde_derive", "serde_json", "io_ipc"]
 io_print = ["comfy-table"]
 # the compute kernels. Disabling this significantly reduces compile time.
 compute_aggregate = ["multiversion"]
@@ -313,6 +314,10 @@ harness = false
 name = "write_json"
 harness = false
 
+[[bench]]
+name = "read_json"
+harness = false
+
 [[bench]]
 name = "slices_iterator"
 harness = false
diff --git a/benches/read_json.rs b/benches/read_json.rs
new file mode 100644
index 0000000000..5d0400c592
--- /dev/null
+++ b/benches/read_json.rs
@@ -0,0 +1,67 @@
+use arrow2::array::Array;
+use arrow2::datatypes::DataType;
+use criterion::{criterion_group, criterion_main, Criterion};
+
+use arrow2::io::json::{read, write};
+use arrow2::util::bench_util::*;
+
+fn prep(array: impl Array + 'static) -> (Vec<u8>, DataType) {
+    let mut data = vec![];
+    let blocks = write::Serializer::new(
+        vec![Ok(Box::new(array) as Box<dyn Array>)].into_iter(),
+        vec![],
+    );
+    // the operation of writing is IO-bounded.
+    write::write(&mut data, blocks).unwrap();
+
+    let value = read::json_deserializer::parse(&data).unwrap();
+
+    let dt = read::infer(&value).unwrap();
+    (data, dt)
+}
+
+fn bench_read(data: &[u8], dt: &DataType) {
+    let value = read::json_deserializer::parse(data).unwrap();
+    read::deserialize(&value, dt.clone()).unwrap();
+}
+
+fn add_benchmark(c: &mut Criterion) {
+    (10..=20).step_by(2).for_each(|log2_size| {
+        let size = 2usize.pow(log2_size);
+
+        let array = create_primitive_array::<i32>(size, 0.1);
+
+        let (data, dt) = prep(array);
+
+        c.bench_function(&format!("read i32 2^{}", log2_size), |b| {
+            b.iter(|| bench_read(&data, &dt))
+        });
+
+        let array = create_primitive_array::<f64>(size, 0.1);
+
+        let (data, dt) = prep(array);
+
+        c.bench_function(&format!("read f64 2^{}", log2_size), |b| {
+            b.iter(|| bench_read(&data, &dt))
+        });
+
+        let array = create_string_array::<i32>(size, 10, 0.1, 42);
+
+        let (data, dt) = prep(array);
+
+        c.bench_function(&format!("read utf8 2^{}", log2_size), |b| {
+            b.iter(|| bench_read(&data, &dt))
+        });
+
+        let array = create_boolean_array(size, 0.1, 0.1);
+
+        let (data, dt) = prep(array);
+
+        c.bench_function(&format!("read bool 2^{}", log2_size), |b| {
+            b.iter(|| bench_read(&data, &dt))
+        });
+    })
+}
+
+criterion_group!(benches, add_benchmark);
+criterion_main!(benches);
diff --git a/examples/json_read.rs b/examples/json_read.rs
index 1630aa6504..608980a62d 100644
--- a/examples/json_read.rs
+++ b/examples/json_read.rs
@@ -1,5 +1,5 @@
-use std::fs::File;
-use std::io::BufReader;
+/// Example of reading a JSON file.
+use std::fs;
 use std::sync::Arc;
 
 use arrow2::array::Array;
@@ -7,12 +7,16 @@ use arrow2::error::Result;
 use arrow2::io::json::read;
 
 fn read_path(path: &str) -> Result<Arc<dyn Array>> {
-    // Example of reading a JSON file.
-    let reader = BufReader::new(File::open(path)?);
-    let json = serde_json::from_reader(reader)?;
+    // read the file into memory (IO-bounded)
+    let data = fs::read(path)?;
 
+    // create a non-owning struct of the data (CPU-bounded)
+    let json = read::json_deserializer::parse(&data)?;
+
+    // use it to infer an Arrow schema (CPU-bounded)
     let data_type = read::infer(&json)?;
 
+    // and deserialize it (CPU-bounded)
     read::deserialize(&json, data_type)
 }
 
diff --git a/parquet_integration/write_parquet.py b/parquet_integration/write_parquet.py
index 3971ff3e54..7f296aed3f 100644
--- a/parquet_integration/write_parquet.py
+++ b/parquet_integration/write_parquet.py
@@ -261,6 +261,32 @@ def case_nested_edge():
     )
 
 
+def case_map() -> Tuple[dict, pa.Schema, str]:
+    s1 = ["a1", "a2"]
+    s2 = ["b1", "b2"]
+    schema = pa.schema(
+        [
+            pa.field(
+                "map",
+                pa.map_(pa.string(), pa.string()),
+            ),
+            pa.field(
+                "map_nullable",
+                pa.map_(pa.string(), pa.string()),
+            ),
+        ]
+    )
+
+    return (
+        {
+            "map": pa.MapArray.from_arrays([0, 2], s1, s2),
+            "map_nullable": pa.MapArray.from_arrays([0, 2], s1, ["b1", None]),
+        },
+        schema,
+        f"map_required_10.parquet",
+    )
+
+
 def write_pyarrow(
     case,
     page_version: int,
@@ -299,7 +325,7 @@ def write_pyarrow(
     )
 
 
-for case in [case_basic_nullable, case_basic_required, case_nested, case_struct, case_nested_edge]:
+for case in [case_basic_nullable, case_basic_required, case_nested, case_struct, case_nested_edge, case_map]:
     for version in [1, 2]:
         for use_dict in [True, False]:
             for compression in ["lz4", None, "snappy"]:
diff --git a/src/array/binary/from.rs b/src/array/binary/from.rs
index c4993a9582..aa575ccd9c 100644
--- a/src/array/binary/from.rs
+++ b/src/array/binary/from.rs
@@ -1,88 +1,12 @@
 use std::iter::FromIterator;
 
-use crate::{array::Offset, trusted_len::TrustedLen};
+use crate::array::Offset;
 
 use super::{BinaryArray, MutableBinaryArray};
 
-impl<O: Offset> BinaryArray<O> {
-    /// Creates a new [`BinaryArray`] from slices of `&[u8]`.
-    pub fn from_slice<T: AsRef<[u8]>, P: AsRef<[T]>>(slice: P) -> Self {
-        Self::from_trusted_len_values_iter(slice.as_ref().iter())
-    }
-
-    /// Creates a new [`BinaryArray`] from a slice of optional `&[u8]`.
-    // Note: this can't be `impl From` because Rust does not allow double `AsRef` on it.
-    pub fn from<T: AsRef<[u8]>, P: AsRef<[Option<T>]>>(slice: P) -> Self {
-        Self::from_trusted_len_iter(slice.as_ref().iter().map(|x| x.as_ref()))
-    }
-
-    /// Creates a [`BinaryArray`] from an iterator of trusted length.
-    #[inline]
-    pub fn from_trusted_len_values_iter<T: AsRef<[u8]>, I: TrustedLen<Item = T>>(
-        iterator: I,
-    ) -> Self {
-        MutableBinaryArray::<O>::from_trusted_len_values_iter(iterator).into()
-    }
-
-    /// Creates a new [`BinaryArray`] from a [`Iterator`] of `&str`.
-    pub fn from_iter_values<T: AsRef<[u8]>, I: Iterator<Item = T>>(iterator: I) -> Self {
-        MutableBinaryArray::<O>::from_iter_values(iterator).into()
-    }
-}
-
 impl<O: Offset, P: AsRef<[u8]>> FromIterator<Option<P>> for BinaryArray<O> {
     #[inline]
     fn from_iter<I: IntoIterator<Item = Option<P>>>(iter: I) -> Self {
         MutableBinaryArray::<O>::from_iter(iter).into()
     }
 }
-
-impl<O: Offset> BinaryArray<O> {
-    /// Creates a [`BinaryArray`] from an iterator of trusted length.
-    /// # Safety
-    /// The iterator must be [`TrustedLen`](https://doc.rust-lang.org/std/iter/trait.TrustedLen.html).
-    /// I.e. that `size_hint().1` correctly reports its length.
-    #[inline]
-    pub unsafe fn from_trusted_len_iter_unchecked<I, P>(iterator: I) -> Self
-    where
-        P: AsRef<[u8]>,
-        I: Iterator<Item = Option<P>>,
-    {
-        MutableBinaryArray::<O>::from_trusted_len_iter_unchecked(iterator).into()
-    }
-
-    /// Creates a [`BinaryArray`] from an iterator of trusted length.
-    #[inline]
-    pub fn from_trusted_len_iter<I, P>(iterator: I) -> Self
-    where
-        P: AsRef<[u8]>,
-        I: TrustedLen<Item = Option<P>>,
-    {
-        // soundness: I is `TrustedLen`
-        unsafe { Self::from_trusted_len_iter_unchecked(iterator) }
-    }
-
-    /// Creates a [`BinaryArray`] from an falible iterator of trusted length.
-    /// # Safety
-    /// The iterator must be [`TrustedLen`](https://doc.rust-lang.org/std/iter/trait.TrustedLen.html).
-    /// I.e. that `size_hint().1` correctly reports its length.
-    #[inline]
-    pub unsafe fn try_from_trusted_len_iter_unchecked<E, I, P>(iterator: I) -> Result<Self, E>
-    where
-        P: AsRef<[u8]>,
-        I: IntoIterator<Item = Result<Option<P>, E>>,
-    {
-        MutableBinaryArray::<O>::try_from_trusted_len_iter_unchecked(iterator).map(|x| x.into())
-    }
-
-    /// Creates a [`BinaryArray`] from an fallible iterator of trusted length.
-    #[inline]
-    pub fn try_from_trusted_len_iter<E, I, P>(iter: I) -> Result<Self, E>
-    where
-        P: AsRef<[u8]>,
-        I: TrustedLen<Item = Result<Option<P>, E>>,
-    {
-        // soundness: I: TrustedLen
-        unsafe { Self::try_from_trusted_len_iter_unchecked(iter) }
-    }
-}
diff --git a/src/array/binary/iterator.rs b/src/array/binary/iterator.rs
index 3dc085ab22..10f0e8a43d 100644
--- a/src/array/binary/iterator.rs
+++ b/src/array/binary/iterator.rs
@@ -1,8 +1,4 @@
-use crate::{
-    array::Offset,
-    bitmap::utils::{zip_validity, ZipValidity},
-    trusted_len::TrustedLen,
-};
+use crate::{array::Offset, bitmap::utils::ZipValidity, trusted_len::TrustedLen};
 
 use super::BinaryArray;
 
@@ -50,19 +46,4 @@ impl<'a, O: Offset> IntoIterator for &'a BinaryArray<O> {
     }
 }
 
-impl<'a, O: Offset> BinaryArray<O> {
-    /// Returns an iterator of `Option<&[u8]>`
-    pub fn iter(&'a self) -> ZipValidity<'a, &'a [u8], BinaryValueIter<'a, O>> {
-        zip_validity(
-            BinaryValueIter::new(self),
-            self.validity.as_ref().map(|x| x.iter()),
-        )
-    }
-
-    /// Returns an iterator of `&[u8]`
-    pub fn values_iter(&'a self) -> BinaryValueIter<'a, O> {
-        BinaryValueIter::new(self)
-    }
-}
-
 unsafe impl<O: Offset> TrustedLen for BinaryValueIter<'_, O> {}
diff --git a/src/array/binary/mod.rs b/src/array/binary/mod.rs
index 9f3c5dad67..ff6fb609ab 100644
--- a/src/array/binary/mod.rs
+++ b/src/array/binary/mod.rs
@@ -1,8 +1,12 @@
 use crate::{
-    bitmap::Bitmap,
+    bitmap::{
+        utils::{zip_validity, ZipValidity},
+        Bitmap,
+    },
     buffer::Buffer,
     datatypes::DataType,
-    error::{Error, Result},
+    error::Error,
+    trusted_len::TrustedLen,
 };
 
 use super::{
@@ -18,7 +22,32 @@ mod from;
 mod mutable;
 pub use mutable::*;
 
-/// A [`BinaryArray`] is a nullable array of bytes - the Arrow equivalent of `Vec<Option<Vec<u8>>>`.
+/// A [`BinaryArray`] is Arrow's semantically equivalent of an immutable `Vec<Option<Vec<u8>>>`.
+/// It implements [`Array`].
+///
+/// The size of this struct is `O(1)`, as all data is stored behind an [`std::sync::Arc`].
+/// # Example
+/// ```
+/// use arrow2::array::BinaryArray;
+/// use arrow2::bitmap::Bitmap;
+/// use arrow2::buffer::Buffer;
+///
+/// let array = BinaryArray::<i32>::from([Some([1, 2].as_ref()), None, Some([3].as_ref())]);
+/// assert_eq!(array.value(0), &[1, 2]);
+/// assert_eq!(array.iter().collect::<Vec<_>>(), vec![Some([1, 2].as_ref()), None, Some([3].as_ref())]);
+/// assert_eq!(array.values_iter().collect::<Vec<_>>(), vec![[1, 2].as_ref(), &[], &[3]]);
+/// // the underlying representation:
+/// assert_eq!(array.values(), &Buffer::from(vec![1, 2, 3]));
+/// assert_eq!(array.offsets(), &Buffer::from(vec![0, 2, 2, 3]));
+/// assert_eq!(array.validity(), Some(&Bitmap::from([true, false, true])));
+/// ```
+///
+/// # Generic parameter
+/// The generic parameter [`Offset`] can only be `i32` or `i64` and tradeoffs maximum array length with
+/// memory usage:
+/// * the sum of lengths of all elements cannot exceed `Offset::MAX`
+/// * the total size of the underlying data is `array.len() * size_of::<Offset>() + sum of lengths of all elements`
+///
 /// # Safety
 /// The following invariants hold:
 /// * Two consecutives `offsets` casted (`as`) to `usize` are valid slices of `values`.
@@ -31,9 +60,8 @@ pub struct BinaryArray<O: Offset> {
     validity: Option<Bitmap>,
 }
 
-// constructors
 impl<O: Offset> BinaryArray<O> {
-    /// Creates a new [`BinaryArray`].
+    /// Returns a [`BinaryArray`] created from its internal representation.
     ///
     /// # Errors
     /// This function returns an error iff:
@@ -48,7 +76,7 @@ impl<O: Offset> BinaryArray<O> {
         offsets: Buffer<O>,
         values: Buffer<u8>,
         validity: Option<Bitmap>,
-    ) -> Result<Self> {
+    ) -> Result<Self, Error> {
         try_check_offsets(&offsets, values.len())?;
 
         if validity
@@ -74,31 +102,134 @@ impl<O: Offset> BinaryArray<O> {
         })
     }
 
-    /// Creates a new [`BinaryArray`].
+    /// Creates a new [`BinaryArray`] from slices of `&[u8]`.
+    pub fn from_slice<T: AsRef<[u8]>, P: AsRef<[T]>>(slice: P) -> Self {
+        Self::from_trusted_len_values_iter(slice.as_ref().iter())
+    }
+
+    /// Creates a new [`BinaryArray`] from a slice of optional `&[u8]`.
+    // Note: this can't be `impl From` because Rust does not allow double `AsRef` on it.
+    pub fn from<T: AsRef<[u8]>, P: AsRef<[Option<T>]>>(slice: P) -> Self {
+        Self::from_trusted_len_iter(slice.as_ref().iter().map(|x| x.as_ref()))
+    }
+
+    /// Returns an iterator of `Option<&[u8]>` over every element of this array.
+    pub fn iter(&self) -> ZipValidity<&[u8], BinaryValueIter<O>> {
+        zip_validity(self.values_iter(), self.validity.as_ref().map(|x| x.iter()))
+    }
+
+    /// Returns an iterator of `&[u8]` over every element of this array, ignoring the validity
+    pub fn values_iter(&self) -> BinaryValueIter<O> {
+        BinaryValueIter::new(self)
+    }
+
+    /// Returns the length of this array
+    #[inline]
+    pub fn len(&self) -> usize {
+        self.offsets.len() - 1
+    }
+
+    /// Returns the element at index `i`
     /// # Panics
-    /// * the offsets are not monotonically increasing
-    /// * The last offset is not equal to the values' length.
-    /// * the validity's length is not equal to `offsets.len() - 1`.
-    /// * The `data_type`'s [`crate::datatypes::PhysicalType`] is not equal to either `Binary` or `LargeBinary`.
+    /// iff `i >= self.len()`
+    #[inline]
+    pub fn value(&self, i: usize) -> &[u8] {
+        assert!(i < self.len());
+        unsafe { self.value_unchecked(i) }
+    }
+
+    /// Returns the element at index `i`
+    /// # Safety
+    /// Assumes that the `i < self.len`.
+    #[inline]
+    pub unsafe fn value_unchecked(&self, i: usize) -> &[u8] {
+        // soundness: the invariant of the function
+        let start = self.offsets.get_unchecked(i).to_usize();
+        let end = self.offsets.get_unchecked(i + 1).to_usize();
+
+        // soundness: the invariant of the struct
+        self.values.get_unchecked(start..end)
+    }
+
+    /// Returns the [`DataType`] of this array.
+    #[inline]
+    pub fn data_type(&self) -> &DataType {
+        &self.data_type
+    }
+
+    /// Returns the values of this [`Utf8Array`].
+    #[inline]
+    pub fn values(&self) -> &Buffer<u8> {
+        &self.values
+    }
+
+    /// Returns the offsets of this [`Utf8Array`].
+    #[inline]
+    pub fn offsets(&self) -> &Buffer<O> {
+        &self.offsets
+    }
+
+    /// The optional validity.
+    #[inline]
+    pub fn validity(&self) -> Option<&Bitmap> {
+        self.validity.as_ref()
+    }
+
+    /// Creates a new [`BinaryArray`] by slicing this [`BinaryArray`].
     /// # Implementation
-    /// This function is `O(N)` - checking monotinicity is `O(N)`
-    pub fn new(
-        data_type: DataType,
-        offsets: Buffer<O>,
-        values: Buffer<u8>,
-        validity: Option<Bitmap>,
-    ) -> Self {
-        Self::try_new(data_type, offsets, values, validity).unwrap()
+    /// This function is `O(1)`: all data will be shared between both arrays.
+    /// # Panics
+    /// iff `offset + length > self.len()`.
+    #[must_use]
+    pub fn slice(&self, offset: usize, length: usize) -> Self {
+        assert!(
+            offset + length <= self.len(),
+            "the offset of the new Buffer cannot exceed the existing length"
+        );
+        unsafe { self.slice_unchecked(offset, length) }
     }
 
-    /// Alias for `new`
-    pub fn from_data(
-        data_type: DataType,
-        offsets: Buffer<O>,
-        values: Buffer<u8>,
-        validity: Option<Bitmap>,
-    ) -> Self {
-        Self::new(data_type, offsets, values, validity)
+    /// Creates a new [`BinaryArray`] by slicing this [`BinaryArray`].
+    /// # Implementation
+    /// This function is `O(1)`: all data will be shared between both arrays.
+    /// # Safety
+    /// The caller must ensure that `offset + length <= self.len()`.
+    #[must_use]
+    pub unsafe fn slice_unchecked(&self, offset: usize, length: usize) -> Self {
+        let validity = self
+            .validity
+            .clone()
+            .map(|x| x.slice_unchecked(offset, length));
+        let offsets = self.offsets.clone().slice_unchecked(offset, length + 1);
+        Self {
+            data_type: self.data_type.clone(),
+            offsets,
+            values: self.values.clone(),
+            validity,
+        }
+    }
+
+    /// Boxes self into a [`Box<dyn Array>`].
+    pub fn boxed(self) -> Box<dyn Array> {
+        Box::new(self)
+    }
+
+    /// Boxes self into a [`std::sync::Arc<dyn Array>`].
+    pub fn arced(self) -> std::sync::Arc<dyn Array> {
+        std::sync::Arc::new(self)
+    }
+
+    /// Clones this [`BinaryArray`] with a different validity.
+    /// # Panic
+    /// Panics iff `validity.len() != self.len()`.
+    #[must_use]
+    pub fn with_validity(&self, validity: Option<Bitmap>) -> Self {
+        if matches!(&validity, Some(bitmap) if bitmap.len() != self.len()) {
+            panic!("validity's length must be equal to the array's length")
+        }
+        let mut arr = self.clone();
+        arr.validity = validity;
+        arr
     }
 
     /// Creates an empty [`BinaryArray`], i.e. whose `.len` is zero.
@@ -131,19 +262,6 @@ impl<O: Offset> BinaryArray<O> {
         }
     }
 
-    /// Boxes self into a [`Box<dyn Array>`].
-    pub fn boxed(self) -> Box<dyn Array> {
-        Box::new(self)
-    }
-
-    /// Boxes self into a [`std::sync::Arc<dyn Array>`].
-    pub fn arced(self) -> std::sync::Arc<dyn Array> {
-        std::sync::Arc::new(self)
-    }
-}
-
-// unsafe constructors
-impl<O: Offset> BinaryArray<O> {
     /// Creates a new [`BinaryArray`] without checking for offsets monotinicity.
     ///
     /// # Errors
@@ -161,7 +279,7 @@ impl<O: Offset> BinaryArray<O> {
         offsets: Buffer<O>,
         values: Buffer<u8>,
         validity: Option<Bitmap>,
-    ) -> Result<Self> {
+    ) -> Result<Self, Error> {
         try_check_offsets_bounds(&offsets, values.len())?;
 
         if validity
@@ -187,138 +305,115 @@ impl<O: Offset> BinaryArray<O> {
         })
     }
 
-    /// Creates a new [`BinaryArray`] without checking for offsets monotinicity.
-    ///
-    /// # Panics
-    /// This function returns an error iff:
-    /// * The last offset is not equal to the values' length.
-    /// * the validity's length is not equal to `offsets.len() - 1`.
-    /// * The `data_type`'s [`crate::datatypes::PhysicalType`] is not equal to either `Binary` or `LargeBinary`.
-    /// # Safety
-    /// This function is unsafe iff:
-    /// * the offsets are not monotonically increasing
-    /// # Implementation
-    /// This function is `O(1)`
-    pub unsafe fn new_unchecked(
+    /// Alias for unwrapping [`Self::try_new`]
+    pub fn new(
         data_type: DataType,
         offsets: Buffer<O>,
         values: Buffer<u8>,
         validity: Option<Bitmap>,
     ) -> Self {
-        Self::try_new_unchecked(data_type, offsets, values, validity).unwrap()
+        Self::try_new(data_type, offsets, values, validity).unwrap()
     }
 
-    /// Alias for [`Self::new_unchecked`]
+    /// Alias for unwrapping [`Self::try_new_unchecked`]
     /// # Safety
     /// This function is unsafe iff:
     /// * the offsets are not monotonically increasing
-    pub unsafe fn from_data_unchecked(
+    pub unsafe fn new_unchecked(
         data_type: DataType,
         offsets: Buffer<O>,
         values: Buffer<u8>,
         validity: Option<Bitmap>,
     ) -> Self {
-        Self::new_unchecked(data_type, offsets, values, validity)
-    }
-}
-
-// must use
-impl<O: Offset> BinaryArray<O> {
-    /// Creates a new [`BinaryArray`] by slicing this [`BinaryArray`].
-    /// # Implementation
-    /// This function is `O(1)`: all data will be shared between both arrays.
-    /// # Panics
-    /// iff `offset + length > self.len()`.
-    #[must_use]
-    pub fn slice(&self, offset: usize, length: usize) -> Self {
-        assert!(
-            offset + length <= self.len(),
-            "the offset of the new Buffer cannot exceed the existing length"
-        );
-        unsafe { self.slice_unchecked(offset, length) }
+        Self::try_new_unchecked(data_type, offsets, values, validity).unwrap()
     }
 
-    /// Creates a new [`BinaryArray`] by slicing this [`BinaryArray`].
-    /// # Implementation
-    /// This function is `O(1)`: all data will be shared between both arrays.
-    /// # Safety
-    /// The caller must ensure that `offset + length <= self.len()`.
-    #[must_use]
-    pub unsafe fn slice_unchecked(&self, offset: usize, length: usize) -> Self {
-        let validity = self
-            .validity
-            .clone()
-            .map(|x| x.slice_unchecked(offset, length));
-        let offsets = self.offsets.clone().slice_unchecked(offset, length + 1);
-        Self {
-            data_type: self.data_type.clone(),
-            offsets,
-            values: self.values.clone(),
-            validity,
-        }
+    /// Returns a [`BinaryArray`] from an iterator of trusted length.
+    ///
+    /// The [`BinaryArray`] is guaranteed to not have a validity
+    #[inline]
+    pub fn from_trusted_len_values_iter<T: AsRef<[u8]>, I: TrustedLen<Item = T>>(
+        iterator: I,
+    ) -> Self {
+        MutableBinaryArray::<O>::from_trusted_len_values_iter(iterator).into()
     }
 
-    /// Clones this [`BinaryArray`] with a different validity.
-    /// # Panic
-    /// Panics iff `validity.len() != self.len()`.
-    #[must_use]
-    pub fn with_validity(&self, validity: Option<Bitmap>) -> Self {
-        if matches!(&validity, Some(bitmap) if bitmap.len() != self.len()) {
-            panic!("validity's length must be equal to the array's length")
-        }
-        let mut arr = self.clone();
-        arr.validity = validity;
-        arr
+    /// Returns a new [`BinaryArray`] from a [`Iterator`] of `&[u8]`.
+    ///
+    /// The [`BinaryArray`] is guaranteed to not have a validity
+    pub fn from_iter_values<T: AsRef<[u8]>, I: Iterator<Item = T>>(iterator: I) -> Self {
+        MutableBinaryArray::<O>::from_iter_values(iterator).into()
     }
-}
 
-// accessors
-impl<O: Offset> BinaryArray<O> {
-    /// Returns the length of this array
+    /// Creates a [`BinaryArray`] from an iterator of trusted length.
+    /// # Safety
+    /// The iterator must be [`TrustedLen`](https://doc.rust-lang.org/std/iter/trait.TrustedLen.html).
+    /// I.e. that `size_hint().1` correctly reports its length.
     #[inline]
-    pub fn len(&self) -> usize {
-        self.offsets.len() - 1
+    pub unsafe fn from_trusted_len_iter_unchecked<I, P>(iterator: I) -> Self
+    where
+        P: AsRef<[u8]>,
+        I: Iterator<Item = Option<P>>,
+    {
+        MutableBinaryArray::<O>::from_trusted_len_iter_unchecked(iterator).into()
     }
 
-    /// Returns the element at index `i`
-    /// # Panics
-    /// iff `i >= self.len()`
-    pub fn value(&self, i: usize) -> &[u8] {
-        let start = self.offsets[i].to_usize();
-        let end = self.offsets[i + 1].to_usize();
-
-        // soundness: the invariant of the struct
-        unsafe { self.values.get_unchecked(start..end) }
+    /// Creates a [`BinaryArray`] from a [`TrustedLen`]
+    #[inline]
+    pub fn from_trusted_len_iter<I, P>(iterator: I) -> Self
+    where
+        P: AsRef<[u8]>,
+        I: TrustedLen<Item = Option<P>>,
+    {
+        // soundness: I is `TrustedLen`
+        unsafe { Self::from_trusted_len_iter_unchecked(iterator) }
     }
 
-    /// Returns the element at index `i`
+    /// Creates a [`BinaryArray`] from an falible iterator of trusted length.
     /// # Safety
-    /// Assumes that the `i < self.len`.
-    pub unsafe fn value_unchecked(&self, i: usize) -> &[u8] {
-        // soundness: the invariant of the function
-        let start = self.offsets.get_unchecked(i).to_usize();
-        let end = self.offsets.get_unchecked(i + 1).to_usize();
-
-        // soundness: the invariant of the struct
-        self.values.get_unchecked(start..end)
+    /// The iterator must be [`TrustedLen`](https://doc.rust-lang.org/std/iter/trait.TrustedLen.html).
+    /// I.e. that `size_hint().1` correctly reports its length.
+    #[inline]
+    pub unsafe fn try_from_trusted_len_iter_unchecked<E, I, P>(iterator: I) -> Result<Self, E>
+    where
+        P: AsRef<[u8]>,
+        I: IntoIterator<Item = Result<Option<P>, E>>,
+    {
+        MutableBinaryArray::<O>::try_from_trusted_len_iter_unchecked(iterator).map(|x| x.into())
     }
 
-    /// The optional validity.
+    /// Creates a [`BinaryArray`] from an fallible iterator of trusted length.
     #[inline]
-    pub fn validity(&self) -> Option<&Bitmap> {
-        self.validity.as_ref()
+    pub fn try_from_trusted_len_iter<E, I, P>(iter: I) -> Result<Self, E>
+    where
+        P: AsRef<[u8]>,
+        I: TrustedLen<Item = Result<Option<P>, E>>,
+    {
+        // soundness: I: TrustedLen
+        unsafe { Self::try_from_trusted_len_iter_unchecked(iter) }
     }
 
-    /// Returns the offsets that slice `.values()` to return valid values.
-    #[inline]
-    pub fn offsets(&self) -> &Buffer<O> {
-        &self.offsets
+    /// Alias for [`Self::new_unchecked`]
+    /// # Safety
+    /// This function is unsafe iff:
+    /// * the offsets are not monotonically increasing
+    pub unsafe fn from_data_unchecked(
+        data_type: DataType,
+        offsets: Buffer<O>,
+        values: Buffer<u8>,
+        validity: Option<Bitmap>,
+    ) -> Self {
+        Self::new_unchecked(data_type, offsets, values, validity)
     }
 
-    /// Returns all values in this array. Use `.offsets()` to slice them.
-    #[inline]
-    pub fn values(&self) -> &Buffer<u8> {
-        &self.values
+    /// Alias for `new`
+    pub fn from_data(
+        data_type: DataType,
+        offsets: Buffer<O>,
+        values: Buffer<u8>,
+        validity: Option<Bitmap>,
+    ) -> Self {
+        Self::new(data_type, offsets, values, validity)
     }
 }
 
diff --git a/src/array/equal/binary.rs b/src/array/equal/binary.rs
index 036f069349..1c86fab6dc 100644
--- a/src/array/equal/binary.rs
+++ b/src/array/equal/binary.rs
@@ -1,4 +1,4 @@
-use crate::array::{Array, BinaryArray, Offset};
+use crate::array::{BinaryArray, Offset};
 
 pub(super) fn equal<O: Offset>(lhs: &BinaryArray<O>, rhs: &BinaryArray<O>) -> bool {
     lhs.data_type() == rhs.data_type() && lhs.len() == rhs.len() && lhs.iter().eq(rhs.iter())
diff --git a/src/array/equal/mod.rs b/src/array/equal/mod.rs
index e916aaa0bb..932ccd67bb 100644
--- a/src/array/equal/mod.rs
+++ b/src/array/equal/mod.rs
@@ -165,6 +165,18 @@ impl PartialEq<&dyn Array> for UnionArray {
     }
 }
 
+impl PartialEq<MapArray> for MapArray {
+    fn eq(&self, other: &Self) -> bool {
+        map::equal(self, other)
+    }
+}
+
+impl PartialEq<&dyn Array> for MapArray {
+    fn eq(&self, other: &&dyn Array) -> bool {
+        equal(self, *other)
+    }
+}
+
 /// Logically compares two [`Array`]s.
 /// Two arrays are logically equal if and only if:
 /// * their data types are equal
diff --git a/src/array/map/fmt.rs b/src/array/map/fmt.rs
new file mode 100644
index 0000000000..60abf56e18
--- /dev/null
+++ b/src/array/map/fmt.rs
@@ -0,0 +1,24 @@
+use std::fmt::{Debug, Formatter, Result, Write};
+
+use super::super::fmt::{get_display, write_vec};
+use super::MapArray;
+
+pub fn write_value<W: Write>(
+    array: &MapArray,
+    index: usize,
+    null: &'static str,
+    f: &mut W,
+) -> Result {
+    let values = array.value(index);
+    let writer = |f: &mut W, index| get_display(values.as_ref(), null)(f, index);
+    write_vec(f, writer, None, values.len(), null, false)
+}
+
+impl Debug for MapArray {
+    fn fmt(&self, f: &mut Formatter<'_>) -> Result {
+        let writer = |f: &mut Formatter, index| write_value(self, index, "None", f);
+
+        write!(f, "MapArray")?;
+        write_vec(f, writer, self.validity.as_ref(), self.len(), "None", false)
+    }
+}
diff --git a/src/array/map/mod.rs b/src/array/map/mod.rs
index 4c7e4754fd..f8b754d618 100644
--- a/src/array/map/mod.rs
+++ b/src/array/map/mod.rs
@@ -10,11 +10,12 @@ use crate::{
 use super::{new_empty_array, specification::try_check_offsets, Array};
 
 mod ffi;
+mod fmt;
 mod iterator;
 pub use iterator::*;
 
 /// An array representing a (key, value), both of arbitrary logical types.
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct MapArray {
     data_type: DataType,
     // invariant: field.len() == offsets.len() - 1
diff --git a/src/array/mod.rs b/src/array/mod.rs
index 0be802df2f..2a8806f078 100644
--- a/src/array/mod.rs
+++ b/src/array/mod.rs
@@ -240,7 +240,7 @@ impl std::fmt::Debug for dyn Array + '_ {
                     fmt_dyn!(self, DictionaryArray::<$T>, f)
                 })
             }
-            Map => todo!(),
+            Map => fmt_dyn!(self, MapArray, f),
         }
     }
 }
diff --git a/src/array/utf8/mod.rs b/src/array/utf8/mod.rs
index 088e77076f..5d89fe4aab 100644
--- a/src/array/utf8/mod.rs
+++ b/src/array/utf8/mod.rs
@@ -42,6 +42,13 @@ pub use mutable::*;
 /// assert_eq!(array.offsets(), &Buffer::from(vec![0, 2, 2, 2 + 5]));
 /// # }
 /// ```
+///
+/// # Generic parameter
+/// The generic parameter [`Offset`] can only be `i32` or `i64` and tradeoffs maximum array length with
+/// memory usage:
+/// * the sum of lengths of all elements cannot exceed `Offset::MAX`
+/// * the total size of the underlying data is `array.len() * size_of::<Offset>() + sum of lengths of all elements`
+///
 /// # Safety
 /// The following invariants hold:
 /// * Two consecutives `offsets` casted (`as`) to `usize` are valid slices of `values`.
@@ -57,7 +64,7 @@ pub struct Utf8Array<O: Offset> {
 
 // constructors
 impl<O: Offset> Utf8Array<O> {
-    /// Returns a [`Utf8Array`] from its internal representation.
+    /// Returns a [`Utf8Array`] created from its internal representation.
     ///
     /// # Errors
     /// This function returns an error iff:
@@ -115,10 +122,7 @@ impl<O: Offset> Utf8Array<O> {
 
     /// Returns an iterator of `Option<&str>`
     pub fn iter(&self) -> ZipValidity<&str, Utf8ValuesIter<O>> {
-        zip_validity(
-            Utf8ValuesIter::new(self),
-            self.validity.as_ref().map(|x| x.iter()),
-        )
+        zip_validity(self.values_iter(), self.validity.as_ref().map(|x| x.iter()))
     }
 
     /// Returns an iterator of `&str`
diff --git a/src/ffi/schema.rs b/src/ffi/schema.rs
index fa7156766a..4f4fc5b8be 100644
--- a/src/ffi/schema.rs
+++ b/src/ffi/schema.rs
@@ -306,7 +306,6 @@ unsafe fn to_data_type(schema: &ArrowSchema) -> Result<DataType> {
                     let size = size_raw
                         .parse::<usize>()
                         .map_err(|_| Error::OutOfSpec("size is not a valid integer".to_string()))?;
-                    println!("schema: {}", size);
                     DataType::FixedSizeBinary(size)
                 }
                 ["+w", size_raw] => {
diff --git a/src/io/avro/read/decompress.rs b/src/io/avro/read/decompress.rs
index 80b8153529..05f5fd43d9 100644
--- a/src/io/avro/read/decompress.rs
+++ b/src/io/avro/read/decompress.rs
@@ -9,6 +9,7 @@ use super::super::{Block, CompressedBlock};
 use super::BlockStreamIterator;
 use super::Compression;
 
+#[cfg(feature = "io_avro_compression")]
 const CRC_TABLE: crc::Crc<u32> = crc::Crc::<u32>::new(&crc::CRC_32_ISO_HDLC);
 
 /// Decompresses an Avro block.
diff --git a/src/io/avro/write/compress.rs b/src/io/avro/write/compress.rs
index 19b69460ee..5121003a2b 100644
--- a/src/io/avro/write/compress.rs
+++ b/src/io/avro/write/compress.rs
@@ -5,6 +5,7 @@ use crate::error::Result;
 use super::Compression;
 use super::{Block, CompressedBlock};
 
+#[cfg(feature = "io_avro_compression")]
 const CRC_TABLE: crc::Crc<u32> = crc::Crc::<u32>::new(&crc::CRC_32_ISO_HDLC);
 
 /// Compresses a [`Block`] to a [`CompressedBlock`].
diff --git a/src/io/json/mod.rs b/src/io/json/mod.rs
index ad406f4ffd..ebbdc92b69 100644
--- a/src/io/json/mod.rs
+++ b/src/io/json/mod.rs
@@ -5,8 +5,8 @@ pub mod write;
 
 use crate::error::Error;
 
-impl From<serde_json::error::Error> for Error {
-    fn from(error: serde_json::error::Error) -> Self {
-        Error::External("".to_string(), Box::new(error))
+impl From<json_deserializer::Error> for Error {
+    fn from(error: json_deserializer::Error) -> Self {
+        Error::ExternalFormat(error.to_string())
     }
 }
diff --git a/src/io/json/read/deserialize.rs b/src/io/json/read/deserialize.rs
index fbee18ab70..41d69f00c0 100644
--- a/src/io/json/read/deserialize.rs
+++ b/src/io/json/read/deserialize.rs
@@ -4,8 +4,7 @@ use std::{collections::hash_map::DefaultHasher, sync::Arc};
 
 use hash_hasher::HashedMap;
 use indexmap::map::IndexMap as HashMap;
-use num_traits::NumCast;
-use serde_json::Value;
+use json_deserializer::{Number, Value};
 
 use crate::{
     array::*,
@@ -17,7 +16,7 @@ use crate::{
 
 /// A function that converts a &Value into an optional tuple of a byte slice and a Value.
 /// This is used to create a dictionary, where the hashing depends on the DataType of the child object.
-type Extract = Box<dyn Fn(&Value) -> Option<(u64, &Value)>>;
+type Extract<'a> = Box<dyn Fn(&'a Value<'a>) -> Option<(u64, &'a Value<'a>)>>;
 
 fn build_extract(data_type: &DataType) -> Extract {
     match data_type {
@@ -27,11 +26,10 @@ fn build_extract(data_type: &DataType) -> Extract {
                 hasher.write(v.as_bytes());
                 Some((hasher.finish(), value))
             }
-            Value::Number(v) => v.as_f64().map(|x| {
-                let mut hasher = DefaultHasher::new();
-                hasher.write(&x.to_le_bytes());
-                (hasher.finish(), value)
-            }),
+            Value::Number(v) => match v {
+                Number::Float(_, _) => todo!(),
+                Number::Integer(_, _) => todo!(),
+            },
             Value::Bool(v) => {
                 let mut hasher = DefaultHasher::new();
                 hasher.write(&[*v as u8]);
@@ -40,25 +38,24 @@ fn build_extract(data_type: &DataType) -> Extract {
             _ => None,
         }),
         DataType::Int32 | DataType::Int64 | DataType::Int16 | DataType::Int8 => {
-            Box::new(move |value| match &value {
-                Value::Number(v) => v.as_f64().map(|x| {
+            Box::new(move |value| {
+                let integer = match value {
+                    Value::Number(number) => Some(deserialize_int_single::<i64>(*number)),
+                    Value::Bool(number) => Some(if *number { 1i64 } else { 0i64 }),
+                    _ => None,
+                };
+                integer.map(|integer| {
                     let mut hasher = DefaultHasher::new();
-                    hasher.write(&x.to_le_bytes());
+                    hasher.write(&integer.to_le_bytes());
                     (hasher.finish(), value)
-                }),
-                Value::Bool(v) => {
-                    let mut hasher = DefaultHasher::new();
-                    hasher.write(&[*v as u8]);
-                    Some((hasher.finish(), value))
-                }
-                _ => None,
+                })
             })
         }
         _ => Box::new(|_| None),
     }
 }
 
-fn deserialize_boolean<A: Borrow<Value>>(rows: &[A]) -> BooleanArray {
+fn deserialize_boolean<'a, A: Borrow<Value<'a>>>(rows: &[A]) -> BooleanArray {
     let iter = rows.iter().map(|row| match row.borrow() {
         Value::Bool(v) => Some(v),
         _ => None,
@@ -66,31 +63,125 @@ fn deserialize_boolean<A: Borrow<Value>>(rows: &[A]) -> BooleanArray {
     BooleanArray::from_trusted_len_iter(iter)
 }
 
-fn deserialize_int<T: NativeType + NumCast, A: Borrow<Value>>(
+fn deserialize_int_single<T>(number: Number) -> T
+where
+    T: NativeType + lexical_core::FromLexical + Pow10,
+{
+    match number {
+        Number::Float(fraction, exponent) => {
+            let integer = fraction.split(|x| *x == b'.').next().unwrap();
+            let mut integer: T = lexical_core::parse(integer).unwrap();
+            if !exponent.is_empty() {
+                let exponent: u32 = lexical_core::parse(exponent).unwrap();
+                integer = integer.pow10(exponent);
+            }
+            integer
+        }
+        Number::Integer(integer, exponent) => {
+            let mut integer: T = lexical_core::parse(integer).unwrap();
+            if !exponent.is_empty() {
+                let exponent: u32 = lexical_core::parse(exponent).unwrap();
+                integer = integer.pow10(exponent);
+            }
+            integer
+        }
+    }
+}
+
+trait Powi10: NativeType + num_traits::One + std::ops::Add {
+    fn powi10(self, exp: i32) -> Self;
+}
+
+impl Powi10 for f32 {
+    #[inline]
+    fn powi10(self, exp: i32) -> Self {
+        self * 10.0f32.powi(exp)
+    }
+}
+
+impl Powi10 for f64 {
+    #[inline]
+    fn powi10(self, exp: i32) -> Self {
+        self * 10.0f64.powi(exp)
+    }
+}
+
+trait Pow10: NativeType + num_traits::One + std::ops::Add {
+    fn pow10(self, exp: u32) -> Self;
+}
+
+macro_rules! impl_pow10 {
+    ($ty:ty) => {
+        impl Pow10 for $ty {
+            #[inline]
+            fn pow10(self, exp: u32) -> Self {
+                self * (10 as $ty).pow(exp)
+            }
+        }
+    };
+}
+impl_pow10!(u8);
+impl_pow10!(u16);
+impl_pow10!(u32);
+impl_pow10!(u64);
+impl_pow10!(i8);
+impl_pow10!(i16);
+impl_pow10!(i32);
+impl_pow10!(i64);
+
+fn deserialize_float_single<T>(number: &Number) -> T
+where
+    T: NativeType + lexical_core::FromLexical + Powi10,
+{
+    match number {
+        Number::Float(float, exponent) => {
+            let mut float: T = lexical_core::parse(float).unwrap();
+            if !exponent.is_empty() {
+                let exponent: i32 = lexical_core::parse(exponent).unwrap();
+                float = float.powi10(exponent);
+            }
+            float
+        }
+        Number::Integer(integer, exponent) => {
+            let mut float: T = lexical_core::parse(integer).unwrap();
+            if !exponent.is_empty() {
+                let exponent: i32 = lexical_core::parse(exponent).unwrap();
+                float = float.powi10(exponent);
+            }
+            float
+        }
+    }
+}
+
+fn deserialize_int<'a, T: NativeType + lexical_core::FromLexical + Pow10, A: Borrow<Value<'a>>>(
     rows: &[A],
     data_type: DataType,
 ) -> PrimitiveArray<T> {
     let iter = rows.iter().map(|row| match row.borrow() {
-        Value::Number(number) => number.as_i64().and_then(num_traits::cast::<i64, T>),
-        Value::Bool(number) => num_traits::cast::<i32, T>(*number as i32),
+        Value::Number(number) => Some(deserialize_int_single(*number)),
+        Value::Bool(number) => Some(if *number { T::one() } else { T::default() }),
         _ => None,
     });
     PrimitiveArray::from_trusted_len_iter(iter).to(data_type)
 }
 
-fn deserialize_float<T: NativeType + NumCast, A: Borrow<Value>>(
+fn deserialize_float<
+    'a,
+    T: NativeType + lexical_core::FromLexical + Powi10,
+    A: Borrow<Value<'a>>,
+>(
     rows: &[A],
     data_type: DataType,
 ) -> PrimitiveArray<T> {
     let iter = rows.iter().map(|row| match row.borrow() {
-        Value::Number(number) => number.as_f64().and_then(num_traits::cast::<f64, T>),
-        Value::Bool(number) => num_traits::cast::<i32, T>(*number as i32),
+        Value::Number(number) => Some(deserialize_float_single(number)),
+        Value::Bool(number) => Some(if *number { T::one() } else { T::default() }),
         _ => None,
     });
     PrimitiveArray::from_trusted_len_iter(iter).to(data_type)
 }
 
-fn deserialize_binary<O: Offset, A: Borrow<Value>>(rows: &[A]) -> BinaryArray<O> {
+fn deserialize_binary<'a, O: Offset, A: Borrow<Value<'a>>>(rows: &[A]) -> BinaryArray<O> {
     let iter = rows.iter().map(|row| match row.borrow() {
         Value::String(v) => Some(v.as_bytes()),
         _ => None,
@@ -98,17 +189,31 @@ fn deserialize_binary<O: Offset, A: Borrow<Value>>(rows: &[A]) -> BinaryArray<O>
     BinaryArray::from_trusted_len_iter(iter)
 }
 
-fn deserialize_utf8<O: Offset, A: Borrow<Value>>(rows: &[A]) -> Utf8Array<O> {
-    let iter = rows.iter().map(|row| match row.borrow() {
-        Value::String(v) => Some(v.clone()),
-        Value::Number(v) => Some(v.to_string()),
-        Value::Bool(v) => Some(v.to_string()),
-        _ => None,
-    });
-    Utf8Array::<O>::from_trusted_len_iter(iter)
+fn deserialize_utf8<'a, O: Offset, A: Borrow<Value<'a>>>(rows: &[A]) -> Utf8Array<O> {
+    let mut array = MutableUtf8Array::<O>::with_capacity(rows.len());
+    let mut scratch = vec![];
+    for row in rows {
+        match row.borrow() {
+            Value::String(v) => array.push(Some(v.as_ref())),
+            Value::Number(number) => match number {
+                Number::Integer(number, exponent) | Number::Float(number, exponent) => {
+                    scratch.clear();
+                    scratch.extend_from_slice(*number);
+                    scratch.push(b'e');
+                    scratch.extend_from_slice(*exponent);
+                }
+            },
+            Value::Bool(v) => array.push(Some(if *v { "true" } else { "false" })),
+            _ => array.push_null(),
+        }
+    }
+    array.into()
 }
 
-fn deserialize_list<O: Offset, A: Borrow<Value>>(rows: &[A], data_type: DataType) -> ListArray<O> {
+fn deserialize_list<'a, O: Offset, A: Borrow<Value<'a>>>(
+    rows: &[A],
+    data_type: DataType,
+) -> ListArray<O> {
     let child = ListArray::<O>::get_child_type(&data_type);
 
     let mut validity = MutableBitmap::with_capacity(rows.len());
@@ -138,7 +243,7 @@ fn deserialize_list<O: Offset, A: Borrow<Value>>(rows: &[A], data_type: DataType
     ListArray::<O>::new(data_type, offsets.into(), values, validity.into())
 }
 
-fn deserialize_struct<A: Borrow<Value>>(rows: &[A], data_type: DataType) -> StructArray {
+fn deserialize_struct<'a, A: Borrow<Value<'a>>>(rows: &[A], data_type: DataType) -> StructArray {
     let fields = StructArray::get_fields(&data_type);
 
     let mut values = fields
@@ -173,7 +278,7 @@ fn deserialize_struct<A: Borrow<Value>>(rows: &[A], data_type: DataType) -> Stru
     StructArray::new(data_type, values, validity.into())
 }
 
-fn deserialize_dictionary<K: DictionaryKey, A: Borrow<Value>>(
+fn deserialize_dictionary<'a, K: DictionaryKey, A: Borrow<Value<'a>>>(
     rows: &[A],
     data_type: DataType,
 ) -> DictionaryArray<K> {
@@ -206,7 +311,10 @@ fn deserialize_dictionary<K: DictionaryKey, A: Borrow<Value>>(
     DictionaryArray::<K>::from_data(keys, values)
 }
 
-pub(crate) fn _deserialize<A: Borrow<Value>>(rows: &[A], data_type: DataType) -> Arc<dyn Array> {
+pub(crate) fn _deserialize<'a, A: Borrow<Value<'a>>>(
+    rows: &[A],
+    data_type: DataType,
+) -> Arc<dyn Array> {
     match &data_type {
         DataType::Null => Arc::new(NullArray::new(data_type, rows.len())),
         DataType::Boolean => Arc::new(deserialize_boolean(rows)),
diff --git a/src/io/json/read/infer_schema.rs b/src/io/json/read/infer_schema.rs
index 336264c431..2ea4f84058 100644
--- a/src/io/json/read/infer_schema.rs
+++ b/src/io/json/read/infer_schema.rs
@@ -1,8 +1,9 @@
 use std::borrow::Borrow;
+use std::collections::BTreeMap;
 
 use indexmap::map::IndexMap as HashMap;
 use indexmap::set::IndexSet as HashSet;
-use serde_json::Value;
+use json_deserializer::{Number, Value};
 
 use crate::datatypes::*;
 use crate::error::Result;
@@ -29,7 +30,7 @@ fn filter_map_nulls(dt: DataType) -> Option<DataType> {
     }
 }
 
-fn infer_object(inner: &serde_json::Map<String, Value>) -> Result<DataType> {
+fn infer_object(inner: &BTreeMap<String, Value>) -> Result<DataType> {
     let fields = inner
         .iter()
         .filter_map(|(key, value)| {
@@ -69,11 +70,10 @@ fn infer_array(values: &[Value]) -> Result<DataType> {
     })
 }
 
-fn infer_number(n: &serde_json::Number) -> DataType {
-    if n.is_f64() {
-        DataType::Float64
-    } else {
-        DataType::Int64
+fn infer_number(n: &Number) -> DataType {
+    match n {
+        Number::Float(..) => DataType::Float64,
+        Number::Integer(..) => DataType::Int64,
     }
 }
 
diff --git a/src/io/json/read/mod.rs b/src/io/json/read/mod.rs
index 908da67b51..ed1ad17f10 100644
--- a/src/io/json/read/mod.rs
+++ b/src/io/json/read/mod.rs
@@ -6,3 +6,5 @@ pub(crate) use deserialize::_deserialize;
 pub use deserialize::deserialize;
 pub(crate) use infer_schema::coerce_data_type;
 pub use infer_schema::infer;
+
+pub use json_deserializer;
diff --git a/src/io/json/write/mod.rs b/src/io/json/write/mod.rs
index 8a416819f7..2278a636f2 100644
--- a/src/io/json/write/mod.rs
+++ b/src/io/json/write/mod.rs
@@ -1,5 +1,6 @@
 //! APIs to write to JSON
 mod serialize;
+mod utf8;
 
 pub use fallible_streaming_iterator::*;
 pub(crate) use serialize::new_serializer;
diff --git a/src/io/json/write/serialize.rs b/src/io/json/write/serialize.rs
index fb90c17ad0..afaf7dab04 100644
--- a/src/io/json/write/serialize.rs
+++ b/src/io/json/write/serialize.rs
@@ -13,6 +13,8 @@ use crate::temporal_conversions::{
 use crate::util::lexical_to_bytes_mut;
 use crate::{array::*, datatypes::DataType, types::NativeType};
 
+use super::utf8;
+
 fn boolean_serializer<'a>(
     array: &'a BooleanArray,
 ) -> Box<dyn StreamingIterator<Item = [u8]> + 'a + Send + Sync> {
@@ -73,7 +75,7 @@ fn utf8_serializer<'a, O: Offset>(
         array.iter(),
         |x, buf| {
             if let Some(x) = x {
-                serde_json::to_writer(buf, x).unwrap();
+                utf8::write_str(buf, x).unwrap();
             } else {
                 buf.extend_from_slice(b"null")
             }
@@ -256,7 +258,7 @@ fn serialize_item(buffer: &mut Vec<u8>, record: &[(&str, &[u8])], is_first_row:
             buffer.push(b',');
         }
         first_item = false;
-        serde_json::to_writer(&mut *buffer, key).unwrap();
+        utf8::write_str(buffer, key).unwrap();
         buffer.push(b':');
         buffer.extend(*value);
     }
diff --git a/src/io/json/write/utf8.rs b/src/io/json/write/utf8.rs
new file mode 100644
index 0000000000..b8c9852217
--- /dev/null
+++ b/src/io/json/write/utf8.rs
@@ -0,0 +1,138 @@
+// Adapted from https://github.com/serde-rs/json/blob/f901012df66811354cb1d490ad59480d8fdf77b5/src/ser.rs
+use std::io;
+
+pub fn write_str<W>(writer: &mut W, value: &str) -> io::Result<()>
+where
+    W: io::Write,
+{
+    writer.write_all(b"\"")?;
+    let bytes = value.as_bytes();
+
+    let mut start = 0;
+
+    for (i, &byte) in bytes.iter().enumerate() {
+        let escape = ESCAPE[byte as usize];
+        if escape == 0 {
+            continue;
+        }
+
+        if start < i {
+            writer.write_all(&bytes[start..i])?;
+        }
+
+        let char_escape = CharEscape::from_escape_table(escape, byte);
+        write_char_escape(writer, char_escape)?;
+
+        start = i + 1;
+    }
+
+    if start != bytes.len() {
+        writer.write_all(&bytes[start..])?;
+    }
+    writer.write_all(b"\"")
+}
+
+const BB: u8 = b'b'; // \x08
+const TT: u8 = b't'; // \x09
+const NN: u8 = b'n'; // \x0A
+const FF: u8 = b'f'; // \x0C
+const RR: u8 = b'r'; // \x0D
+const QU: u8 = b'"'; // \x22
+const BS: u8 = b'\\'; // \x5C
+const UU: u8 = b'u'; // \x00...\x1F except the ones above
+const __: u8 = 0;
+
+// Lookup table of escape sequences. A value of b'x' at index i means that byte
+// i is escaped as "\x" in JSON. A value of 0 means that byte i is not escaped.
+static ESCAPE: [u8; 256] = [
+    //   1   2   3   4   5   6   7   8   9   A   B   C   D   E   F
+    UU, UU, UU, UU, UU, UU, UU, UU, BB, TT, NN, UU, FF, RR, UU, UU, // 0
+    UU, UU, UU, UU, UU, UU, UU, UU, UU, UU, UU, UU, UU, UU, UU, UU, // 1
+    __, __, QU, __, __, __, __, __, __, __, __, __, __, __, __, __, // 2
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // 3
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // 4
+    __, __, __, __, __, __, __, __, __, __, __, __, BS, __, __, __, // 5
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // 6
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // 7
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // 8
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // 9
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // A
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // B
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // C
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // D
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // E
+    __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // F
+];
+
+/// Represents a character escape code in a type-safe manner.
+pub enum CharEscape {
+    /// An escaped quote `"`
+    Quote,
+    /// An escaped reverse solidus `\`
+    ReverseSolidus,
+    // An escaped solidus `/`
+    //Solidus,
+    /// An escaped backspace character (usually escaped as `\b`)
+    Backspace,
+    /// An escaped form feed character (usually escaped as `\f`)
+    FormFeed,
+    /// An escaped line feed character (usually escaped as `\n`)
+    LineFeed,
+    /// An escaped carriage return character (usually escaped as `\r`)
+    CarriageReturn,
+    /// An escaped tab character (usually escaped as `\t`)
+    Tab,
+    /// An escaped ASCII plane control character (usually escaped as
+    /// `\u00XX` where `XX` are two hex characters)
+    AsciiControl(u8),
+}
+
+impl CharEscape {
+    #[inline]
+    fn from_escape_table(escape: u8, byte: u8) -> CharEscape {
+        match escape {
+            self::BB => CharEscape::Backspace,
+            self::TT => CharEscape::Tab,
+            self::NN => CharEscape::LineFeed,
+            self::FF => CharEscape::FormFeed,
+            self::RR => CharEscape::CarriageReturn,
+            self::QU => CharEscape::Quote,
+            self::BS => CharEscape::ReverseSolidus,
+            self::UU => CharEscape::AsciiControl(byte),
+            _ => unreachable!(),
+        }
+    }
+}
+
+#[inline]
+fn write_char_escape<W>(writer: &mut W, char_escape: CharEscape) -> io::Result<()>
+where
+    W: io::Write,
+{
+    use self::CharEscape::*;
+
+    let s = match char_escape {
+        Quote => b"\\\"",
+        ReverseSolidus => b"\\\\",
+        //Solidus => b"\\/",
+        Backspace => b"\\b",
+        FormFeed => b"\\f",
+        LineFeed => b"\\n",
+        CarriageReturn => b"\\r",
+        Tab => b"\\t",
+        AsciiControl(byte) => {
+            static HEX_DIGITS: [u8; 16] = *b"0123456789abcdef";
+            let bytes = &[
+                b'\\',
+                b'u',
+                b'0',
+                b'0',
+                HEX_DIGITS[(byte >> 4) as usize],
+                HEX_DIGITS[(byte & 0xF) as usize],
+            ];
+            return writer.write_all(bytes);
+        }
+    };
+
+    writer.write_all(s)
+}
diff --git a/src/io/json_integration/mod.rs b/src/io/json_integration/mod.rs
index 279f2ec55f..dc522c60c8 100644
--- a/src/io/json_integration/mod.rs
+++ b/src/io/json_integration/mod.rs
@@ -5,6 +5,8 @@
 use serde_derive::{Deserialize, Serialize};
 use serde_json::Value;
 
+use crate::error::Error;
+
 pub mod read;
 pub mod write;
 
@@ -117,3 +119,9 @@ pub struct ArrowJsonColumn {
     /// the children
     pub children: Option<Vec<ArrowJsonColumn>>,
 }
+
+impl From<serde_json::Error> for Error {
+    fn from(error: serde_json::Error) -> Self {
+        Error::ExternalFormat(error.to_string())
+    }
+}
diff --git a/src/io/ndjson/read/deserialize.rs b/src/io/ndjson/read/deserialize.rs
index 49e1ef6492..dae87d66d7 100644
--- a/src/io/ndjson/read/deserialize.rs
+++ b/src/io/ndjson/read/deserialize.rs
@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use serde_json::Value;
+use json_deserializer::parse;
 
 use crate::array::Array;
 use crate::datatypes::DataType;
@@ -22,7 +22,7 @@ pub fn deserialize(rows: &[String], data_type: DataType) -> Result<Arc<dyn Array
         ));
     }
 
-    deserialize_iter(rows.iter(), data_type)
+    deserialize_iter(rows.iter().map(|x| x.as_ref()), data_type)
 }
 
 /// Deserializes an iterator of rows into an [`Array`] of [`DataType`].
@@ -31,14 +31,14 @@ pub fn deserialize(rows: &[String], data_type: DataType) -> Result<Arc<dyn Array
 /// This function is guaranteed to return an array of length equal to the leng
 /// # Errors
 /// This function errors iff any of the rows is not a valid JSON (i.e. the format is not valid NDJSON).
-pub fn deserialize_iter<A: AsRef<str>>(
-    rows: impl Iterator<Item = A>,
+pub fn deserialize_iter<'a>(
+    rows: impl Iterator<Item = &'a str>,
     data_type: DataType,
 ) -> Result<Arc<dyn Array>, Error> {
     // deserialize strings to `Value`s
     let rows = rows
-        .map(|row| serde_json::from_str(row.as_ref()).map_err(Error::from))
-        .collect::<Result<Vec<Value>, Error>>()?;
+        .map(|row| parse(row.as_bytes()).map_err(Error::from))
+        .collect::<Result<Vec<_>, Error>>()?;
 
     // deserialize &[Value] to Array
     Ok(_deserialize(&rows, data_type))
diff --git a/src/io/ndjson/read/file.rs b/src/io/ndjson/read/file.rs
index 2b0174a02b..6ed5497489 100644
--- a/src/io/ndjson/read/file.rs
+++ b/src/io/ndjson/read/file.rs
@@ -2,8 +2,7 @@ use std::io::BufRead;
 
 use fallible_streaming_iterator::FallibleStreamingIterator;
 use indexmap::set::IndexSet as HashSet;
-use serde_json;
-use serde_json::Value;
+use json_deserializer::parse;
 
 use crate::{
     datatypes::DataType,
@@ -115,7 +114,7 @@ pub fn infer<R: std::io::BufRead>(
 
     let mut data_types = HashSet::new();
     while let Some(rows) = reader.next()? {
-        let value: Value = serde_json::from_str(&rows[0])?; // 0 because it is row by row
+        let value = parse(rows[0].as_bytes())?; // 0 because it is row by row
         let data_type = infer_json(&value)?;
         if data_type != DataType::Null {
             data_types.insert(data_type);
@@ -134,7 +133,7 @@ pub fn infer<R: std::io::BufRead>(
 pub fn infer_iter<A: AsRef<str>>(rows: impl Iterator<Item = A>) -> Result<DataType> {
     let mut data_types = HashSet::new();
     for row in rows {
-        let v: Value = serde_json::from_str(row.as_ref())?;
+        let v = parse(row.as_ref().as_bytes())?;
         let data_type = infer_json(&v)?;
         if data_type != DataType::Null {
             data_types.insert(data_type);
diff --git a/src/io/parquet/read/deserialize/binary/basic.rs b/src/io/parquet/read/deserialize/binary/basic.rs
index 33772cfcde..0f10f0578d 100644
--- a/src/io/parquet/read/deserialize/binary/basic.rs
+++ b/src/io/parquet/read/deserialize/binary/basic.rs
@@ -68,7 +68,8 @@ pub(super) struct Required<'a> {
 
 impl<'a> Required<'a> {
     pub fn new(page: &'a DataPage) -> Self {
-        let values = SizedBinaryIter::new(page.buffer(), page.num_values());
+        let (_, _, values) = utils::split_buffer(page);
+        let values = SizedBinaryIter::new(values, page.num_values());
 
         Self { values }
     }
diff --git a/src/io/parquet/read/deserialize/binary/nested.rs b/src/io/parquet/read/deserialize/binary/nested.rs
index 9546112555..2cc11e5b95 100644
--- a/src/io/parquet/read/deserialize/binary/nested.rs
+++ b/src/io/parquet/read/deserialize/binary/nested.rs
@@ -120,6 +120,7 @@ impl<'a, O: Offset> utils::Decoder<'a> for BinaryDecoder<O> {
                 }
             }
             State::OptionalDictionary(page_validity, page_values) => {
+                println!("optional_dict");
                 let dict_values = page_values.dict.values();
                 let dict_offsets = page_values.dict.offsets();
 
diff --git a/src/io/parquet/read/deserialize/boolean/basic.rs b/src/io/parquet/read/deserialize/boolean/basic.rs
index 6f6c31b0f2..7ef7a5de89 100644
--- a/src/io/parquet/read/deserialize/boolean/basic.rs
+++ b/src/io/parquet/read/deserialize/boolean/basic.rs
@@ -55,8 +55,9 @@ struct FilteredRequired<'a> {
 
 impl<'a> FilteredRequired<'a> {
     pub fn new(page: &'a DataPage) -> Self {
+        let (_, _, values) = utils::split_buffer(page);
         // todo: replace this by an iterator over slices, for faster deserialization
-        let values = BitmapIter::new(page.buffer(), 0, page.num_values());
+        let values = BitmapIter::new(values, 0, page.num_values());
 
         let rows = get_selected_rows(page);
         let values = SliceFilteredIter::new(values, rows);
diff --git a/src/io/parquet/read/deserialize/boolean/nested.rs b/src/io/parquet/read/deserialize/boolean/nested.rs
index b2e2477fb4..b2d83d57e7 100644
--- a/src/io/parquet/read/deserialize/boolean/nested.rs
+++ b/src/io/parquet/read/deserialize/boolean/nested.rs
@@ -25,8 +25,9 @@ struct Required<'a> {
 
 impl<'a> Required<'a> {
     pub fn new(page: &'a DataPage) -> Self {
+        let (_, _, values) = utils::split_buffer(page);
         Self {
-            values: page.buffer(),
+            values,
             offset: 0,
             length: page.num_values(),
         }
diff --git a/src/io/parquet/read/deserialize/mod.rs b/src/io/parquet/read/deserialize/mod.rs
index cfc5f2034d..3d15791697 100644
--- a/src/io/parquet/read/deserialize/mod.rs
+++ b/src/io/parquet/read/deserialize/mod.rs
@@ -11,9 +11,9 @@ mod struct_;
 mod utils;
 
 use crate::{
-    array::{Array, BinaryArray, FixedSizeListArray, ListArray, Utf8Array},
+    array::{Array, BinaryArray, FixedSizeListArray, ListArray, MapArray, Utf8Array},
     datatypes::{DataType, Field},
-    error::Result,
+    error::{Error, Result},
 };
 
 use self::nested_utils::{InitNested, NestedArrayIter, NestedState};
@@ -300,7 +300,31 @@ where
             let columns = columns.into_iter().rev().collect();
             Box::new(struct_::StructIterator::new(columns, fields.clone()))
         }
-        other => todo!("{other:?}"),
+        Map(inner, _) => {
+            init.push(InitNested::List(field.is_nullable));
+            let iter = columns_to_iter_recursive(
+                columns,
+                types,
+                inner.as_ref().clone(),
+                init,
+                chunk_size,
+            )?;
+            Box::new(iter.map(move |x| {
+                let (nested, inner) = x?;
+                let array = MapArray::new(
+                    field.data_type().clone(),
+                    vec![0, inner.len() as i32].into(),
+                    inner,
+                    None,
+                );
+                Ok((nested, array.arced()))
+            }))
+        }
+        other => {
+            return Err(Error::nyi(format!(
+                "Deserializing type {other:?} from parquet"
+            )))
+        }
     })
 }
 
diff --git a/src/io/parquet/read/file.rs b/src/io/parquet/read/file.rs
index effb5c1c94..f854f5c90f 100644
--- a/src/io/parquet/read/file.rs
+++ b/src/io/parquet/read/file.rs
@@ -115,12 +115,15 @@ impl<R: Read + Seek> FileReader<R> {
     fn next_row_group(&mut self) -> Result<Option<RowGroupDeserializer>> {
         let result = self.row_groups.next().transpose()?;
 
-        self.remaining_rows = self.remaining_rows.saturating_sub(
-            result
-                .as_ref()
-                .map(|x| x.num_rows())
-                .unwrap_or(self.remaining_rows),
-        );
+        // If current_row_group is None, then there will be no elements to remove.
+        if self.current_row_group.is_some() {
+            self.remaining_rows = self.remaining_rows.saturating_sub(
+                result
+                    .as_ref()
+                    .map(|x| x.num_rows())
+                    .unwrap_or(self.remaining_rows),
+            );
+        }
         Ok(result)
     }
 }
diff --git a/src/io/parquet/read/row_group.rs b/src/io/parquet/read/row_group.rs
index b53e60f271..fea9638e5e 100644
--- a/src/io/parquet/read/row_group.rs
+++ b/src/io/parquet/read/row_group.rs
@@ -78,10 +78,12 @@ impl Iterator for RowGroupDeserializer {
             })
             .collect::<Result<Vec<_>>>()
             .map(Chunk::new);
-        self.remaining_rows -= chunk
-            .as_ref()
-            .map(|x| x.len())
-            .unwrap_or(self.remaining_rows);
+        self.remaining_rows = self.remaining_rows.saturating_sub(
+            chunk
+                .as_ref()
+                .map(|x| x.len())
+                .unwrap_or(self.remaining_rows),
+        );
 
         Some(chunk)
     }
diff --git a/src/io/parquet/read/schema/convert.rs b/src/io/parquet/read/schema/convert.rs
index 9e32ee3057..5f23dc71dd 100644
--- a/src/io/parquet/read/schema/convert.rs
+++ b/src/io/parquet/read/schema/convert.rs
@@ -219,6 +219,10 @@ fn non_repeated_group(
     match (logical_type, converted_type) {
         (Some(GroupLogicalType::List), _) => to_list(fields, parent_name),
         (None, Some(GroupConvertedType::List)) => to_list(fields, parent_name),
+        (Some(GroupLogicalType::Map), _) => to_list(fields, parent_name),
+        (None, Some(GroupConvertedType::Map) | Some(GroupConvertedType::MapKeyValue)) => {
+            to_map(fields)
+        }
         _ => to_struct(fields),
     }
 }
@@ -234,6 +238,13 @@ fn to_struct(fields: &[ParquetType]) -> Option<DataType> {
     }
 }
 
+/// Converts a parquet group type to an arrow [`DataType::Struct`].
+/// Returns [`None`] if all its fields are empty
+fn to_map(fields: &[ParquetType]) -> Option<DataType> {
+    let inner = to_field(&fields[0])?;
+    Some(DataType::Map(Box::new(inner), false))
+}
+
 /// Entry point for converting parquet group type.
 ///
 /// This function takes care of logical type and repetition.
diff --git a/src/io/parquet/read/statistics/map.rs b/src/io/parquet/read/statistics/map.rs
new file mode 100644
index 0000000000..ddc09c9c51
--- /dev/null
+++ b/src/io/parquet/read/statistics/map.rs
@@ -0,0 +1,64 @@
+use crate::{
+    array::{Array, MapArray, MutableArray},
+    datatypes::DataType,
+    error::Error,
+};
+
+use super::make_mutable;
+
+#[derive(Debug)]
+pub struct DynMutableMapArray {
+    data_type: DataType,
+    pub inner: Box<dyn MutableArray>,
+}
+
+impl DynMutableMapArray {
+    pub fn try_with_capacity(data_type: DataType, capacity: usize) -> Result<Self, Error> {
+        let inner = match data_type.to_logical_type() {
+            DataType::Map(inner, _) => inner,
+            _ => unreachable!(),
+        };
+        let inner = make_mutable(inner.data_type(), capacity)?;
+
+        Ok(Self { data_type, inner })
+    }
+}
+
+impl MutableArray for DynMutableMapArray {
+    fn data_type(&self) -> &DataType {
+        &self.data_type
+    }
+
+    fn len(&self) -> usize {
+        self.inner.len()
+    }
+
+    fn validity(&self) -> Option<&crate::bitmap::MutableBitmap> {
+        None
+    }
+
+    fn as_box(&mut self) -> Box<dyn Array> {
+        Box::new(MapArray::new(
+            self.data_type.clone(),
+            vec![0, self.inner.len() as i32].into(),
+            self.inner.as_arc(),
+            None,
+        ))
+    }
+
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn as_mut_any(&mut self) -> &mut dyn std::any::Any {
+        self
+    }
+
+    fn push_null(&mut self) {
+        todo!()
+    }
+
+    fn shrink_to_fit(&mut self) {
+        todo!()
+    }
+}
diff --git a/src/io/parquet/read/statistics/mod.rs b/src/io/parquet/read/statistics/mod.rs
index d1b6d9e4b5..4f3eaeff7b 100644
--- a/src/io/parquet/read/statistics/mod.rs
+++ b/src/io/parquet/read/statistics/mod.rs
@@ -22,6 +22,7 @@ mod boolean;
 mod dictionary;
 mod fixlen;
 mod list;
+mod map;
 mod primitive;
 mod struct_;
 mod utf8;
@@ -37,6 +38,8 @@ pub enum Count {
     Single(UInt64Array),
     /// struct arrays have a count as a struct of UInt64
     Struct(StructArray),
+    /// map arrays have a count as a map of UInt64
+    Map(MapArray),
 }
 
 /// Arrow-deserialized parquet Statistics of a file
@@ -76,6 +79,15 @@ impl From<MutableStatistics> for Statistics {
                 .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
                 .null_count
@@ -96,6 +108,15 @@ impl From<MutableStatistics> for Statistics {
                     .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
@@ -151,6 +172,10 @@ fn make_mutable(data_type: &DataType, capacity: usize) -> Result<Box<dyn Mutable
             data_type.clone(),
             capacity,
         )?),
+        PhysicalType::Map => Box::new(map::DynMutableMapArray::try_with_capacity(
+            data_type.clone(),
+            capacity,
+        )?),
         other => {
             return Err(Error::NotYetImplemented(format!(
                 "Deserializing parquet stats from {:?} is still not implemented",
@@ -168,6 +193,11 @@ fn create_dt(data_type: &DataType) -> DataType {
                 .map(|f| Field::new(&f.name, create_dt(&f.data_type), f.is_nullable))
                 .collect(),
         )
+    } else if let DataType::Map(f, ordered) = data_type.to_logical_type() {
+        DataType::Map(
+            Box::new(Field::new(&f.name, create_dt(&f.data_type), f.is_nullable)),
+            *ordered,
+        )
     } else {
         DataType::UInt64
     }
@@ -329,6 +359,31 @@ fn push(
                     )
                 });
         }
+        Map(_, _) => {
+            let min = min
+                .as_mut_any()
+                .downcast_mut::<map::DynMutableMapArray>()
+                .unwrap();
+            let max = max
+                .as_mut_any()
+                .downcast_mut::<map::DynMutableMapArray>()
+                .unwrap();
+            let distinct_count = distinct_count
+                .as_mut_any()
+                .downcast_mut::<map::DynMutableMapArray>()
+                .unwrap();
+            let null_count = null_count
+                .as_mut_any()
+                .downcast_mut::<map::DynMutableMapArray>()
+                .unwrap();
+            return push(
+                stats,
+                min.inner.as_mut(),
+                max.inner.as_mut(),
+                distinct_count.inner.as_mut(),
+                null_count.inner.as_mut(),
+            );
+        }
         _ => {}
     }
 
diff --git a/src/io/parquet/read/statistics/struct_.rs b/src/io/parquet/read/statistics/struct_.rs
index 085737d624..1f59188b5a 100644
--- a/src/io/parquet/read/statistics/struct_.rs
+++ b/src/io/parquet/read/statistics/struct_.rs
@@ -30,7 +30,7 @@ impl MutableArray for DynMutableStructArray {
     }
 
     fn len(&self) -> usize {
-        self.inner.len()
+        self.inner[0].len()
     }
 
     fn validity(&self) -> Option<&crate::bitmap::MutableBitmap> {


Test patch:
diff --git a/tests/it/io/json/read.rs b/tests/it/io/json/read.rs
index db1b067bb4..e584324931 100644
--- a/tests/it/io/json/read.rs
+++ b/tests/it/io/json/read.rs
@@ -7,7 +7,7 @@ use super::*;
 
 #[test]
 fn read_json() -> Result<()> {
-    let data = r#"[
+    let data = br#"[
         {
             "a": 1
         },
@@ -19,7 +19,7 @@ fn read_json() -> Result<()> {
         }
     ]"#;
 
-    let json = serde_json::from_slice(data.as_bytes())?;
+    let json = json_deserializer::parse(data)?;
 
     let data_type = read::infer(&json)?;
 
diff --git a/tests/it/io/ndjson/read.rs b/tests/it/io/ndjson/read.rs
index 66ea5e0aba..ef3e6882e5 100644
--- a/tests/it/io/ndjson/read.rs
+++ b/tests/it/io/ndjson/read.rs
@@ -156,7 +156,7 @@ fn invalid_infer_schema() -> Result<()> {
     let re = ndjson_read::infer(&mut Cursor::new("city,lat,lng"), None);
     assert_eq!(
         re.err().unwrap().to_string(),
-        "External error: expected value at line 1 column 1",
+        "External format error: InvalidToken(99)",
     );
     Ok(())
 }
@@ -249,7 +249,7 @@ fn invalid_read_record() -> Result<()> {
 
     assert_eq!(
         arrays.err().unwrap().to_string(),
-        "External error: expected value at line 1 column 1",
+        "External format error: InvalidToken(99)",
     );
     Ok(())
 }
diff --git a/tests/it/io/parquet/mod.rs b/tests/it/io/parquet/mod.rs
index d60796378c..700f455758 100644
--- a/tests/it/io/parquet/mod.rs
+++ b/tests/it/io/parquet/mod.rs
@@ -790,6 +790,147 @@ pub fn pyarrow_struct_statistics(column: &str) -> Statistics {
     }
 }
 
+pub fn pyarrow_map(column: &str) -> Box<dyn Array> {
+    match column {
+        "map" => {
+            let s1 = [Some("a1"), Some("a2")];
+            let s2 = [Some("b1"), Some("b2")];
+            let dt = DataType::Struct(vec![
+                Field::new("key", DataType::Utf8, false),
+                Field::new("value", DataType::Utf8, true),
+            ]);
+            MapArray::try_new(
+                DataType::Map(Box::new(Field::new("entries", dt.clone(), false)), false),
+                vec![0, 2].into(),
+                StructArray::try_new(
+                    dt,
+                    vec![
+                        Utf8Array::<i32>::from(s1).arced(),
+                        Utf8Array::<i32>::from(s2).arced(),
+                    ],
+                    None,
+                )
+                .unwrap()
+                .arced(),
+                None,
+            )
+            .unwrap()
+            .boxed()
+        }
+        "map_nullable" => {
+            let s1 = [Some("a1"), Some("a2")];
+            let s2 = [Some("b1"), None];
+            let dt = DataType::Struct(vec![
+                Field::new("key", DataType::Utf8, false),
+                Field::new("value", DataType::Utf8, true),
+            ]);
+            MapArray::try_new(
+                DataType::Map(Box::new(Field::new("entries", dt.clone(), false)), false),
+                vec![0, 2].into(),
+                StructArray::try_new(
+                    dt,
+                    vec![
+                        Utf8Array::<i32>::from(s1).arced(),
+                        Utf8Array::<i32>::from(s2).arced(),
+                    ],
+                    None,
+                )
+                .unwrap()
+                .arced(),
+                None,
+            )
+            .unwrap()
+            .boxed()
+        }
+        _ => unreachable!(),
+    }
+}
+
+pub fn pyarrow_map_statistics(column: &str) -> Statistics {
+    let new_map = |arrays: Vec<Arc<dyn Array>>, names: Vec<String>| {
+        let fields = names
+            .into_iter()
+            .zip(arrays.iter())
+            .map(|(n, a)| Field::new(n, a.data_type().clone(), true))
+            .collect::<Vec<_>>();
+        MapArray::new(
+            DataType::Map(
+                Box::new(Field::new("items", DataType::Struct(fields.clone()), false)),
+                false,
+            ),
+            vec![0, arrays[0].len() as i32].into(),
+            StructArray::new(DataType::Struct(fields), arrays, None).arced(),
+            None,
+        )
+    };
+
+    let names = vec!["key".to_string(), "value".to_string()];
+
+    match column {
+        "map" => Statistics {
+            distinct_count: Count::Map(new_map(
+                vec![
+                    Arc::new(UInt64Array::from([None])),
+                    Arc::new(UInt64Array::from([None])),
+                ],
+                names.clone(),
+            )),
+            null_count: Count::Map(new_map(
+                vec![
+                    Arc::new(UInt64Array::from([Some(0)])),
+                    Arc::new(UInt64Array::from([Some(0)])),
+                ],
+                names.clone(),
+            )),
+            min_value: Box::new(new_map(
+                vec![
+                    Arc::new(Utf8Array::<i32>::from_slice(["a1"])),
+                    Arc::new(Utf8Array::<i32>::from_slice(["b1"])),
+                ],
+                names.clone(),
+            )),
+            max_value: Box::new(new_map(
+                vec![
+                    Arc::new(Utf8Array::<i32>::from_slice(["a2"])),
+                    Arc::new(Utf8Array::<i32>::from_slice(["b2"])),
+                ],
+                names,
+            )),
+        },
+        "map_nullable" => Statistics {
+            distinct_count: Count::Map(new_map(
+                vec![
+                    Arc::new(UInt64Array::from([None])),
+                    Arc::new(UInt64Array::from([None])),
+                ],
+                names.clone(),
+            )),
+            null_count: Count::Map(new_map(
+                vec![
+                    Arc::new(UInt64Array::from([Some(0)])),
+                    Arc::new(UInt64Array::from([Some(1)])),
+                ],
+                names.clone(),
+            )),
+            min_value: Box::new(new_map(
+                vec![
+                    Arc::new(Utf8Array::<i32>::from_slice(["a1"])),
+                    Arc::new(Utf8Array::<i32>::from_slice(["b1"])),
+                ],
+                names.clone(),
+            )),
+            max_value: Box::new(new_map(
+                vec![
+                    Arc::new(Utf8Array::<i32>::from_slice(["a2"])),
+                    Arc::new(Utf8Array::<i32>::from_slice(["b1"])),
+                ],
+                names,
+            )),
+        },
+        _ => unreachable!(),
+    }
+}
+
 fn integration_write(schema: &Schema, batches: &[Chunk<Arc<dyn Array>>]) -> Result<Vec<u8>> {
     let options = WriteOptions {
         write_statistics: true,
diff --git a/tests/it/io/parquet/read.rs b/tests/it/io/parquet/read.rs
index 57b6cbf155..4e426ef5ba 100644
--- a/tests/it/io/parquet/read.rs
+++ b/tests/it/io/parquet/read.rs
@@ -38,6 +38,7 @@ fn test_pyarrow_integration(
         ("nested", false) => pyarrow_nested_nullable(column),
         ("nested_edge", false) => pyarrow_nested_edge(column),
         ("struct", false) => pyarrow_struct(column),
+        ("map", true) => pyarrow_map(column),
         _ => unreachable!(),
     };
 
@@ -47,11 +48,13 @@ fn test_pyarrow_integration(
         ("nested", false) => pyarrow_nested_nullable_statistics(column),
         ("nested_edge", false) => pyarrow_nested_edge_statistics(column),
         ("struct", false) => pyarrow_struct_statistics(column),
+        ("map", true) => pyarrow_map_statistics(column),
         _ => unreachable!(),
     };
 
     assert_eq!(expected.as_ref(), array.as_ref());
     if ![
+        // pyarrow outputs an incorrect number of null count for nested types - ARROW-16299
         "list_int16",
         "list_large_binary",
         "list_int64",
@@ -62,10 +65,12 @@ fn test_pyarrow_integration(
         "list_bool",
         "list_nested_inner_required_required_i64",
         "list_nested_inner_required_i64",
+        // pyarrow reports an incorrect min/max for MapArray
+        "map",
+        "map_nullable",
     ]
     .contains(&column)
     {
-        // pyarrow outputs an incorrect number of null count for nested types - ARROW-16299
         assert_eq!(expected_statistics, statistics);
     }
 
@@ -458,6 +463,16 @@ fn v1_nested_edge_2() -> Result<()> {
     test_pyarrow_integration("null", 1, "nested_edge", false, false, None)
 }
 
+#[test]
+fn v1_map() -> Result<()> {
+    test_pyarrow_integration("map", 1, "map", false, true, None)
+}
+
+#[test]
+fn v1_map_nullable() -> Result<()> {
+    test_pyarrow_integration("map_nullable", 1, "map", false, true, None)
+}
+
 #[cfg(feature = "io_parquet_compression")]
 #[test]
 fn all_types() -> Result<()> {