Repository: jorgecarleitao/arrow2
Pull request #1226 (2022-08-15 18:05:18)

Added support to read and write Parquet's delta-bitpacked (integer encoding)
This PR adds support to read `Encoding::DeltaLengthByteArray`, an important encoding for integer types. For now this is only supported on non-nested types.

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

```diff
@@            Coverage Diff             @@
##             main    #1226      +/-   ##
==========================================
+ Coverage   83.21%   83.38%   +0.16%     
==========================================
  Files         358      359       +1     
  Lines       37224    37677     +453     
==========================================
+ Hits        30976    31416     +440     
- Misses       6248     6261      +13     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1226?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/io/parquet/write/dictionary.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1226/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvZGljdGlvbmFyeS5ycw==) | `86.33% <ø> (ø)` | |
| [...c/io/parquet/read/deserialize/primitive/integer.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1226/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvcmVhZC9kZXNlcmlhbGl6ZS9wcmltaXRpdmUvaW50ZWdlci5ycw==) | `87.09% <87.09%> (ø)` | |
| [src/io/parquet/write/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1226/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvbW9kLnJz) | `86.35% <88.46%> (-0.16%)` | :arrow_down: |
| [src/io/parquet/write/primitive/basic.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1226/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvcHJpbWl0aXZlL2Jhc2ljLnJz) | `98.77% <98.85%> (-0.20%)` | :arrow_down: |
| [src/io/parquet/read/deserialize/primitive/basic.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1226/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvcmVhZC9kZXNlcmlhbGl6ZS9wcmltaXRpdmUvYmFzaWMucnM=) | `95.40% <100.00%> (-0.52%)` | :arrow_down: |
| [src/io/parquet/read/deserialize/simple.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1226/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvcmVhZC9kZXNlcmlhbGl6ZS9zaW1wbGUucnM=) | `83.88% <100.00%> (ø)` | |
| [src/io/parquet/write/primitive/nested.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1226/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvcHJpbWl0aXZlL25lc3RlZC5ycw==) | `97.22% <100.00%> (ø)` | |
| [src/io/ipc/read/reader.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1226/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL3JlYWRlci5ycw==) | `94.66% <0.00%> (-1.34%)` | :arrow_down: |
| [src/io/ipc/read/file.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1226/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL2ZpbGUucnM=) | `96.87% <0.00%> (-0.45%)` | :arrow_down: |
| ... and [10 more](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1226/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | |

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


Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 26dd1acc38..e52ad35508 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -74,7 +74,7 @@ futures = { version = "0.3", optional = true }
 async-stream = { version = "0.3.2", optional = true }
 
 # parquet support
-parquet2 = { version = "0.15.0", optional = true, default_features = false, features = ["async"] }
+parquet2 = { version = "0.16", optional = true, default_features = false, features = ["async"] }
 
 # avro support
 avro-schema = { version = "0.3", optional = true }
diff --git a/src/array/mod.rs b/src/array/mod.rs
index 951e14d71a..a4fd5e4362 100644
--- a/src/array/mod.rs
+++ b/src/array/mod.rs
@@ -9,7 +9,7 @@
 //! * [`Utf8Array`] and [`MutableUtf8Array`], an array of variable length utf8 values
 //! * [`BinaryArray`] and [`MutableBinaryArray`], an array of opaque variable length values
 //! * [`ListArray`] and [`MutableListArray`], an array of arrays (e.g. `[[1, 2], None, [], [None]]`)
-//! * [`StructArray`], an array of arrays identified by a string (e.g. `{"a": [1, 2], "b": [true, false]}`)
+//! * [`StructArray`] and [`MutableStructArray`], an array of arrays identified by a string (e.g. `{"a": [1, 2], "b": [true, false]}`)
 //! All immutable arrays implement the trait object [`Array`] and that can be downcasted
 //! to a concrete struct based on [`PhysicalType`](crate::datatypes::PhysicalType) available from [`Array::data_type`].
 //! All immutable arrays are backed by [`Buffer`](crate::buffer::Buffer) and thus cloning and slicing them is `O(1)`.
diff --git a/src/array/struct_/mutable.rs b/src/array/struct_/mutable.rs
index cc02146a77..b35e1064b9 100644
--- a/src/array/struct_/mutable.rs
+++ b/src/array/struct_/mutable.rs
@@ -110,10 +110,7 @@ impl MutableStructArray {
         }
     }
 
-    /// Call this after pushing into each child array.
-    /// # Panics
-    /// This function panics if any of the children does not have exactly one more
-    /// element than before the last call of `push`.
+    /// Call this once for each "row" of children you push.
     pub fn push(&mut self, valid: bool) {
         match &mut self.validity {
             Some(validity) => validity.push(valid),
@@ -122,7 +119,6 @@ impl MutableStructArray {
                 false => self.init_validity(),
             },
         };
-        self.assert_lengths();
     }
 
     fn push_null(&mut self) {
diff --git a/src/io/parquet/mod.rs b/src/io/parquet/mod.rs
index 87090910bb..0468436911 100644
--- a/src/io/parquet/mod.rs
+++ b/src/io/parquet/mod.rs
@@ -22,6 +22,6 @@ impl From<parquet2::error::Error> for Error {
 
 impl From<Error> for parquet2::error::Error {
     fn from(error: Error) -> Self {
-        parquet2::error::Error::General(error.to_string())
+        parquet2::error::Error::OutOfSpec(error.to_string())
     }
 }
diff --git a/src/io/parquet/read/deserialize/binary/basic.rs b/src/io/parquet/read/deserialize/binary/basic.rs
index 545008ad11..8790b43ea4 100644
--- a/src/io/parquet/read/deserialize/binary/basic.rs
+++ b/src/io/parquet/read/deserialize/binary/basic.rs
@@ -13,7 +13,7 @@ use crate::{
     bitmap::{Bitmap, MutableBitmap},
     buffer::Buffer,
     datatypes::DataType,
-    error::Result,
+    error::{Error, Result},
 };
 
 use super::super::utils::{
@@ -51,13 +51,13 @@ impl<'a> Delta<'a> {
     pub fn try_new(page: &'a DataPage) -> Result<Self> {
         let (_, _, values) = split_buffer(page)?;
 
-        let mut lengths_iter = delta_length_byte_array::Decoder::new(values);
+        let mut lengths_iter = delta_length_byte_array::Decoder::try_new(values)?;
 
         #[allow(clippy::needless_collect)] // we need to consume it to get the values
         let lengths = lengths_iter
             .by_ref()
-            .map(|x| x as usize)
-            .collect::<Vec<_>>();
+            .map(|x| x.map(|x| x as usize).map_err(Error::from))
+            .collect::<Result<Vec<_>>>()?;
 
         let values = lengths_iter.into_values();
         Ok(Self {
@@ -405,20 +405,26 @@ impl<'a, O: Offset> utils::Decoder<'a> for BinaryDecoder<O> {
             }
             State::OptionalDictionary(page_validity, page_values) => {
                 let page_dict = &page_values.dict;
-                let op = move |index: u32| page_dict[index as usize].as_ref();
                 utils::extend_from_decoder(
                     validity,
                     page_validity,
                     Some(additional),
                     values,
-                    &mut page_values.values.by_ref().map(op),
+                    &mut page_values
+                        .values
+                        .by_ref()
+                        .map(|index| page_dict[index.unwrap() as usize].as_ref()),
                 )
             }
             State::RequiredDictionary(page) => {
                 let page_dict = &page.dict;
-                let op = move |index: u32| page_dict[index as usize].as_ref();
 
-                for x in page.values.by_ref().map(op).take(additional) {
+                for x in page
+                    .values
+                    .by_ref()
+                    .map(|index| page_dict[index.unwrap() as usize].as_ref())
+                    .take(additional)
+                {
                     values.push(x)
                 }
             }
@@ -442,21 +448,26 @@ impl<'a, O: Offset> utils::Decoder<'a> for BinaryDecoder<O> {
             }
             State::FilteredRequiredDictionary(page) => {
                 let page_dict = &page.dict;
-                let op = move |index: u32| page_dict[index as usize].as_ref();
-
-                for x in page.values.by_ref().map(op).take(additional) {
+                for x in page
+                    .values
+                    .by_ref()
+                    .map(|index| page_dict[index.unwrap() as usize].as_ref())
+                    .take(additional)
+                {
                     values.push(x)
                 }
             }
             State::FilteredOptionalDictionary(page_validity, page_values) => {
                 let page_dict = &page_values.dict;
-                let op = move |index: u32| page_dict[index as usize].as_ref();
                 utils::extend_from_decoder(
                     validity,
                     page_validity,
                     Some(additional),
                     values,
-                    &mut page_values.values.by_ref().map(op),
+                    &mut page_values
+                        .values
+                        .by_ref()
+                        .map(|index| page_dict[index.unwrap() as usize].as_ref()),
                 )
             }
         }
diff --git a/src/io/parquet/read/deserialize/binary/nested.rs b/src/io/parquet/read/deserialize/binary/nested.rs
index 19d362a967..8c0f5ef419 100644
--- a/src/io/parquet/read/deserialize/binary/nested.rs
+++ b/src/io/parquet/read/deserialize/binary/nested.rs
@@ -90,7 +90,7 @@ impl<'a, O: Offset> NestedDecoder<'a> for BinaryDecoder<O> {
         )
     }
 
-    fn push_valid(&self, state: &mut Self::State, decoded: &mut Self::DecodedState) {
+    fn push_valid(&self, state: &mut Self::State, decoded: &mut Self::DecodedState) -> Result<()> {
         let (values, validity) = decoded;
         match state {
             State::Optional(page) => {
@@ -104,18 +104,25 @@ impl<'a, O: Offset> NestedDecoder<'a> for BinaryDecoder<O> {
             }
             State::RequiredDictionary(page) => {
                 let dict_values = &page.dict;
-                let op = move |index: u32| dict_values[index as usize].as_ref();
-                let item = page.values.next().map(op).unwrap_or_default();
+                let item = page
+                    .values
+                    .next()
+                    .map(|index| dict_values[index.unwrap() as usize].as_ref())
+                    .unwrap_or_default();
                 values.push(item);
             }
             State::OptionalDictionary(page) => {
                 let dict_values = &page.dict;
-                let op = move |index: u32| dict_values[index as usize].as_ref();
-                let item = page.values.next().map(op).unwrap_or_default();
+                let item = page
+                    .values
+                    .next()
+                    .map(|index| dict_values[index.unwrap() as usize].as_ref())
+                    .unwrap_or_default();
                 values.push(item);
                 validity.push(true);
             }
         }
+        Ok(())
     }
 
     fn push_null(&self, decoded: &mut Self::DecodedState) {
diff --git a/src/io/parquet/read/deserialize/boolean/nested.rs b/src/io/parquet/read/deserialize/boolean/nested.rs
index d4c3e8ee12..e8a3b0d5c4 100644
--- a/src/io/parquet/read/deserialize/boolean/nested.rs
+++ b/src/io/parquet/read/deserialize/boolean/nested.rs
@@ -82,7 +82,7 @@ impl<'a> NestedDecoder<'a> for BooleanDecoder {
         )
     }
 
-    fn push_valid(&self, state: &mut State, decoded: &mut Self::DecodedState) {
+    fn push_valid(&self, state: &mut State, decoded: &mut Self::DecodedState) -> Result<()> {
         let (values, validity) = decoded;
         match state {
             State::Optional(page_values) => {
@@ -95,6 +95,7 @@ impl<'a> NestedDecoder<'a> for BooleanDecoder {
                 values.push(value);
             }
         }
+        Ok(())
     }
 
     fn push_null(&self, decoded: &mut Self::DecodedState) {
diff --git a/src/io/parquet/read/deserialize/dictionary/mod.rs b/src/io/parquet/read/deserialize/dictionary/mod.rs
index 0f37935908..43fcaa8ab2 100644
--- a/src/io/parquet/read/deserialize/dictionary/mod.rs
+++ b/src/io/parquet/read/deserialize/dictionary/mod.rs
@@ -163,7 +163,8 @@ where
                 Some(remaining),
                 values,
                 &mut page.values.by_ref().map(|x| {
-                    let x: usize = x.try_into().unwrap();
+                    // todo: rm unwrap
+                    let x: usize = x.unwrap().try_into().unwrap();
                     match x.try_into() {
                         Ok(key) => key,
                         // todo: convert this to an error.
@@ -176,7 +177,8 @@ where
                     page.values
                         .by_ref()
                         .map(|x| {
-                            let x: usize = x.try_into().unwrap();
+                            // todo: rm unwrap
+                            let x: usize = x.unwrap().try_into().unwrap();
                             let x: K = match x.try_into() {
                                 Ok(key) => key,
                                 // todo: convert this to an error.
@@ -195,7 +197,8 @@ where
                 Some(remaining),
                 values,
                 &mut page_values.by_ref().map(|x| {
-                    let x: usize = x.try_into().unwrap();
+                    // todo: rm unwrap
+                    let x: usize = x.unwrap().try_into().unwrap();
                     let x: K = match x.try_into() {
                         Ok(key) => key,
                         // todo: convert this to an error.
@@ -211,7 +214,8 @@ where
                     page.values
                         .by_ref()
                         .map(|x| {
-                            let x: usize = x.try_into().unwrap();
+                            // todo: rm unwrap
+                            let x: usize = x.unwrap().try_into().unwrap();
                             let x: K = match x.try_into() {
                                 Ok(key) => key,
                                 // todo: convert this to an error.
diff --git a/src/io/parquet/read/deserialize/dictionary/nested.rs b/src/io/parquet/read/deserialize/dictionary/nested.rs
index 6d01679f25..89b8701231 100644
--- a/src/io/parquet/read/deserialize/dictionary/nested.rs
+++ b/src/io/parquet/read/deserialize/dictionary/nested.rs
@@ -110,11 +110,11 @@ impl<'a, K: DictionaryKey> NestedDecoder<'a> for DictionaryDecoder<K> {
         )
     }
 
-    fn push_valid(&self, state: &mut Self::State, decoded: &mut Self::DecodedState) {
+    fn push_valid(&self, state: &mut Self::State, decoded: &mut Self::DecodedState) -> Result<()> {
         let (values, validity) = decoded;
         match state {
             State::Optional(page_values) => {
-                let key = page_values.next();
+                let key = page_values.next().transpose()?;
                 // todo: convert unwrap to error
                 let key = match K::try_from(key.unwrap_or_default() as usize) {
                     Ok(key) => key,
@@ -124,7 +124,7 @@ impl<'a, K: DictionaryKey> NestedDecoder<'a> for DictionaryDecoder<K> {
                 validity.push(true);
             }
             State::Required(page_values) => {
-                let key = page_values.values.next();
+                let key = page_values.values.next().transpose()?;
                 let key = match K::try_from(key.unwrap_or_default() as usize) {
                     Ok(key) => key,
                     Err(_) => todo!(),
@@ -132,6 +132,7 @@ impl<'a, K: DictionaryKey> NestedDecoder<'a> for DictionaryDecoder<K> {
                 values.push(key);
             }
         }
+        Ok(())
     }
 
     fn push_null(&self, decoded: &mut Self::DecodedState) {
diff --git a/src/io/parquet/read/deserialize/fixed_size_binary/basic.rs b/src/io/parquet/read/deserialize/fixed_size_binary/basic.rs
index e8de6cbf3f..ab47aa98cf 100644
--- a/src/io/parquet/read/deserialize/fixed_size_binary/basic.rs
+++ b/src/io/parquet/read/deserialize/fixed_size_binary/basic.rs
@@ -227,27 +227,26 @@ impl<'a> Decoder<'a> for BinaryDecoder {
                     values.push(x)
                 }
             }
-            State::OptionalDictionary(page) => {
-                let op = |index: u32| {
-                    let index = index as usize;
+            State::OptionalDictionary(page) => extend_from_decoder(
+                validity,
+                &mut page.validity,
+                Some(remaining),
+                values,
+                page.values.by_ref().map(|index| {
+                    let index = index.unwrap() as usize;
                     &page.dict[index * self.size..(index + 1) * self.size]
-                };
-
-                extend_from_decoder(
-                    validity,
-                    &mut page.validity,
-                    Some(remaining),
-                    values,
-                    page.values.by_ref().map(op),
-                )
-            }
+                }),
+            ),
             State::RequiredDictionary(page) => {
-                let op = |index: u32| {
-                    let index = index as usize;
-                    &page.dict[index * self.size..(index + 1) * self.size]
-                };
-
-                for x in page.values.by_ref().map(op).take(remaining) {
+                for x in page
+                    .values
+                    .by_ref()
+                    .map(|index| {
+                        let index = index.unwrap() as usize;
+                        &page.dict[index * self.size..(index + 1) * self.size]
+                    })
+                    .take(remaining)
+                {
                     values.push(x)
                 }
             }
diff --git a/src/io/parquet/read/deserialize/nested_utils.rs b/src/io/parquet/read/deserialize/nested_utils.rs
index 73679d3422..8bfc2e8c95 100644
--- a/src/io/parquet/read/deserialize/nested_utils.rs
+++ b/src/io/parquet/read/deserialize/nested_utils.rs
@@ -258,7 +258,7 @@ pub(super) trait NestedDecoder<'a> {
     /// Initializes a new state
     fn with_capacity(&self, capacity: usize) -> Self::DecodedState;
 
-    fn push_valid(&self, state: &mut Self::State, decoded: &mut Self::DecodedState);
+    fn push_valid(&self, state: &mut Self::State, decoded: &mut Self::DecodedState) -> Result<()>;
     fn push_null(&self, decoded: &mut Self::DecodedState);
 
     fn deserialize_dict(&self, page: &DictPage) -> Self::Dictionary;
@@ -309,9 +309,9 @@ impl<'a> NestedPage<'a> {
         let max_def_level = page.descriptor.max_def_level;
 
         let reps =
-            HybridRleDecoder::new(rep_levels, get_bit_width(max_rep_level), page.num_values());
+            HybridRleDecoder::try_new(rep_levels, get_bit_width(max_rep_level), page.num_values())?;
         let defs =
-            HybridRleDecoder::new(def_levels, get_bit_width(max_def_level), page.num_values());
+            HybridRleDecoder::try_new(def_levels, get_bit_width(max_def_level), page.num_values())?;
 
         let iter = reps.zip(defs).peekable();
 
@@ -377,7 +377,7 @@ pub(super) fn extend<'a, D: NestedDecoder<'a>>(
         &mut decoded,
         decoder,
         additional,
-    );
+    )?;
     *remaining -= nested.len() - existing;
     items.push_back((nested, decoded));
 
@@ -393,7 +393,7 @@ pub(super) fn extend<'a, D: NestedDecoder<'a>>(
             &mut decoded,
             decoder,
             additional,
-        );
+        )?;
         *remaining -= nested.len();
         items.push_back((nested, decoded));
     }
@@ -407,7 +407,7 @@ fn extend_offsets2<'a, D: NestedDecoder<'a>>(
     decoded: &mut D::DecodedState,
     decoder: &D,
     additional: usize,
-) {
+) -> Result<()> {
     let mut values_count = vec![0; nested.len()];
 
     for (depth, nest) in nested.iter().enumerate().skip(1) {
@@ -431,6 +431,8 @@ fn extend_offsets2<'a, D: NestedDecoder<'a>>(
 
     let mut rows = 0;
     while let Some((rep, def)) = page.iter.next() {
+        let rep = rep?;
+        let def = def?;
         if rep == 0 {
             rows += 1;
         }
@@ -455,7 +457,7 @@ fn extend_offsets2<'a, D: NestedDecoder<'a>>(
                     // the leaf / primitive
                     let is_valid = (def != cum_sum[depth]) || !nest.is_nullable();
                     if right_level && is_valid {
-                        decoder.push_valid(values_state, decoded);
+                        decoder.push_valid(values_state, decoded)?;
                     } else {
                         decoder.push_null(decoded);
                     }
@@ -463,12 +465,19 @@ fn extend_offsets2<'a, D: NestedDecoder<'a>>(
             }
         }
 
-        let next_rep = page.iter.peek().map(|x| x.0).unwrap_or(0);
+        let next_rep = *page
+            .iter
+            .peek()
+            .map(|x| x.0.as_ref())
+            .transpose()
+            .unwrap() // todo: fix this
+            .unwrap_or(&0);
 
         if next_rep == 0 && rows == additional {
             break;
         }
     }
+    Ok(())
 }
 
 #[inline]
diff --git a/src/io/parquet/read/deserialize/primitive/basic.rs b/src/io/parquet/read/deserialize/primitive/basic.rs
index 648e78782a..34a3702f3d 100644
--- a/src/io/parquet/read/deserialize/primitive/basic.rs
+++ b/src/io/parquet/read/deserialize/primitive/basic.rs
@@ -19,7 +19,7 @@ use super::super::utils::{get_selected_rows, FilteredOptionalPageValidity, Optio
 use super::super::Pages;
 
 #[derive(Debug)]
-struct FilteredRequiredValues<'a> {
+pub(super) struct FilteredRequiredValues<'a> {
     values: SliceFilteredIter<std::slice::ChunksExact<'a, u8>>,
 }
 
@@ -89,7 +89,7 @@ where
 
 // The state of a `DataPage` of `Primitive` parquet primitive type
 #[derive(Debug)]
-enum State<'a, T>
+pub(super) enum State<'a, T>
 where
     T: NativeType,
 {
@@ -118,7 +118,7 @@ where
 }
 
 #[derive(Debug)]
-struct PrimitiveDecoder<T, P, F>
+pub(super) struct PrimitiveDecoder<T, P, F>
 where
     T: NativeType,
     P: ParquetNativeType,
@@ -126,7 +126,7 @@ where
 {
     phantom: std::marker::PhantomData<T>,
     phantom_p: std::marker::PhantomData<P>,
-    op: F,
+    pub op: F,
 }
 
 impl<T, P, F> PrimitiveDecoder<T, P, F>
@@ -136,7 +136,7 @@ where
     F: Fn(P) -> T,
 {
     #[inline]
-    fn new(op: F) -> Self {
+    pub(super) fn new(op: F) -> Self {
         Self {
             phantom: std::marker::PhantomData,
             phantom_p: std::marker::PhantomData,
@@ -183,9 +183,9 @@ where
                 Ok(State::Optional(validity, values))
             }
             (Encoding::Plain, _, false, false) => Ok(State::Required(Values::try_new::<P>(page)?)),
-            (Encoding::Plain, _, false, true) => Ok(State::FilteredRequired(
-                FilteredRequiredValues::try_new::<P>(page)?,
-            )),
+            (Encoding::Plain, _, false, true) => {
+                FilteredRequiredValues::try_new::<P>(page).map(State::FilteredRequired)
+            }
             (Encoding::Plain, _, true, true) => Ok(State::FilteredOptional(
                 FilteredOptionalPageValidity::try_new(page)?,
                 Values::try_new::<P>(page)?,
@@ -232,12 +232,18 @@ where
                     page_validity,
                     Some(remaining),
                     values,
-                    &mut page_values.values.by_ref().map(op1),
+                    &mut page_values.values.by_ref().map(|x| x.unwrap()).map(op1),
                 )
             }
             State::RequiredDictionary(page) => {
                 let op1 = |index: u32| page.dict[index as usize];
-                values.extend(page.values.by_ref().map(op1).take(remaining));
+                values.extend(
+                    page.values
+                        .by_ref()
+                        .map(|x| x.unwrap())
+                        .map(op1)
+                        .take(remaining),
+                );
             }
             State::FilteredRequired(page) => {
                 values.extend(
diff --git a/src/io/parquet/read/deserialize/primitive/integer.rs b/src/io/parquet/read/deserialize/primitive/integer.rs
new file mode 100644
index 0000000000..f55b0bec3d
--- /dev/null
+++ b/src/io/parquet/read/deserialize/primitive/integer.rs
@@ -0,0 +1,268 @@
+use std::collections::VecDeque;
+
+use num_traits::AsPrimitive;
+use parquet2::{
+    deserialize::SliceFilteredIter,
+    encoding::{delta_bitpacked::Decoder, Encoding},
+    page::{split_buffer, DataPage, DictPage},
+    schema::Repetition,
+    types::NativeType as ParquetNativeType,
+};
+
+use crate::{
+    array::MutablePrimitiveArray,
+    bitmap::MutableBitmap,
+    datatypes::DataType,
+    error::{Error, Result},
+    io::parquet::read::deserialize::utils::{
+        get_selected_rows, FilteredOptionalPageValidity, OptionalPageValidity,
+    },
+    types::NativeType,
+};
+
+use super::super::utils;
+use super::super::Pages;
+
+use super::basic::{finish, PrimitiveDecoder, State as PrimitiveState};
+
+/// The state of a [`DataPage`] of an integer parquet type (i32 or i64)
+#[derive(Debug)]
+enum State<'a, T>
+where
+    T: NativeType,
+{
+    Common(PrimitiveState<'a, T>),
+    DeltaBinaryPackedRequired(Decoder<'a>),
+    DeltaBinaryPackedOptional(OptionalPageValidity<'a>, Decoder<'a>),
+    FilteredDeltaBinaryPackedRequired(SliceFilteredIter<Decoder<'a>>),
+    FilteredDeltaBinaryPackedOptional(FilteredOptionalPageValidity<'a>, Decoder<'a>),
+}
+
+impl<'a, T> utils::PageState<'a> for State<'a, T>
+where
+    T: NativeType,
+{
+    fn len(&self) -> usize {
+        match self {
+            State::Common(state) => state.len(),
+            State::DeltaBinaryPackedRequired(state) => state.size_hint().0,
+            State::DeltaBinaryPackedOptional(state, _) => state.len(),
+            State::FilteredDeltaBinaryPackedRequired(state) => state.size_hint().0,
+            State::FilteredDeltaBinaryPackedOptional(state, _) => state.len(),
+        }
+    }
+}
+
+/// Decoder of integer parquet type
+#[derive(Debug)]
+struct IntDecoder<T, P, F>(PrimitiveDecoder<T, P, F>)
+where
+    T: NativeType,
+    P: ParquetNativeType,
+    i64: num_traits::AsPrimitive<P>,
+    F: Fn(P) -> T;
+
+impl<T, P, F> IntDecoder<T, P, F>
+where
+    T: NativeType,
+    P: ParquetNativeType,
+    i64: num_traits::AsPrimitive<P>,
+    F: Fn(P) -> T,
+{
+    #[inline]
+    fn new(op: F) -> Self {
+        Self(PrimitiveDecoder::new(op))
+    }
+}
+
+impl<'a, T, P, F> utils::Decoder<'a> for IntDecoder<T, P, F>
+where
+    T: NativeType,
+    P: ParquetNativeType,
+    i64: num_traits::AsPrimitive<P>,
+    F: Copy + Fn(P) -> T,
+{
+    type State = State<'a, T>;
+    type Dict = Vec<T>;
+    type DecodedState = (Vec<T>, MutableBitmap);
+
+    fn build_state(&self, page: &'a DataPage, dict: Option<&'a Self::Dict>) -> Result<Self::State> {
+        let is_optional =
+            page.descriptor.primitive_type.field_info.repetition == Repetition::Optional;
+        let is_filtered = page.selected_rows().is_some();
+
+        match (page.encoding(), dict, is_optional, is_filtered) {
+            (Encoding::DeltaBinaryPacked, _, false, false) => {
+                let (_, _, values) = split_buffer(page)?;
+                Decoder::try_new(values)
+                    .map(State::DeltaBinaryPackedRequired)
+                    .map_err(Error::from)
+            }
+            (Encoding::DeltaBinaryPacked, _, true, false) => {
+                let (_, _, values) = split_buffer(page)?;
+                Ok(State::DeltaBinaryPackedOptional(
+                    OptionalPageValidity::try_new(page)?,
+                    Decoder::try_new(values)?,
+                ))
+            }
+            (Encoding::DeltaBinaryPacked, _, false, true) => {
+                let (_, _, values) = split_buffer(page)?;
+                let values = Decoder::try_new(values)?;
+
+                let rows = get_selected_rows(page);
+                let values = SliceFilteredIter::new(values, rows);
+
+                Ok(State::FilteredDeltaBinaryPackedRequired(values))
+            }
+            (Encoding::DeltaBinaryPacked, _, true, true) => {
+                let (_, _, values) = split_buffer(page)?;
+                let values = Decoder::try_new(values)?;
+
+                Ok(State::FilteredDeltaBinaryPackedOptional(
+                    FilteredOptionalPageValidity::try_new(page)?,
+                    values,
+                ))
+            }
+            _ => self.0.build_state(page, dict).map(State::Common),
+        }
+    }
+
+    fn with_capacity(&self, capacity: usize) -> Self::DecodedState {
+        self.0.with_capacity(capacity)
+    }
+
+    fn extend_from_state(
+        &self,
+        state: &mut Self::State,
+        decoded: &mut Self::DecodedState,
+        remaining: usize,
+    ) {
+        let (values, validity) = decoded;
+        match state {
+            State::Common(state) => self.0.extend_from_state(state, decoded, remaining),
+            State::DeltaBinaryPackedRequired(state) => {
+                values.extend(
+                    state
+                        .by_ref()
+                        .map(|x| x.unwrap().as_())
+                        .map(self.0.op)
+                        .take(remaining),
+                );
+            }
+            State::DeltaBinaryPackedOptional(page_validity, page_values) => {
+                utils::extend_from_decoder(
+                    validity,
+                    page_validity,
+                    Some(remaining),
+                    values,
+                    page_values
+                        .by_ref()
+                        .map(|x| x.unwrap().as_())
+                        .map(self.0.op),
+                )
+            }
+            State::FilteredDeltaBinaryPackedRequired(page) => {
+                values.extend(
+                    page.by_ref()
+                        .map(|x| x.unwrap().as_())
+                        .map(self.0.op)
+                        .take(remaining),
+                );
+            }
+            State::FilteredDeltaBinaryPackedOptional(page_validity, page_values) => {
+                utils::extend_from_decoder(
+                    validity,
+                    page_validity,
+                    Some(remaining),
+                    values,
+                    page_values
+                        .by_ref()
+                        .map(|x| x.unwrap().as_())
+                        .map(self.0.op),
+                );
+            }
+        }
+    }
+
+    fn deserialize_dict(&self, page: &DictPage) -> Self::Dict {
+        self.0.deserialize_dict(page)
+    }
+}
+
+/// An [`Iterator`] adapter over [`Pages`] assumed to be encoded as primitive arrays
+/// encoded as parquet integer types
+#[derive(Debug)]
+pub struct IntegerIter<T, I, P, F>
+where
+    I: Pages,
+    T: NativeType,
+    P: ParquetNativeType,
+    F: Fn(P) -> T,
+{
+    iter: I,
+    data_type: DataType,
+    items: VecDeque<(Vec<T>, MutableBitmap)>,
+    remaining: usize,
+    chunk_size: Option<usize>,
+    dict: Option<Vec<T>>,
+    op: F,
+    phantom: std::marker::PhantomData<P>,
+}
+
+impl<T, I, P, F> IntegerIter<T, I, P, F>
+where
+    I: Pages,
+    T: NativeType,
+
+    P: ParquetNativeType,
+    F: Copy + Fn(P) -> T,
+{
+    pub fn new(
+        iter: I,
+        data_type: DataType,
+        num_rows: usize,
+        chunk_size: Option<usize>,
+        op: F,
+    ) -> Self {
+        Self {
+            iter,
+            data_type,
+            items: VecDeque::new(),
+            dict: None,
+            remaining: num_rows,
+            chunk_size,
+            op,
+            phantom: Default::default(),
+        }
+    }
+}
+
+impl<T, I, P, F> Iterator for IntegerIter<T, I, P, F>
+where
+    I: Pages,
+    T: NativeType,
+    P: ParquetNativeType,
+    i64: num_traits::AsPrimitive<P>,
+    F: Copy + Fn(P) -> T,
+{
+    type Item = Result<MutablePrimitiveArray<T>>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let maybe_state = utils::next(
+            &mut self.iter,
+            &mut self.items,
+            &mut self.dict,
+            &mut self.remaining,
+            self.chunk_size,
+            &IntDecoder::new(self.op),
+        );
+        match maybe_state {
+            utils::MaybeNext::Some(Ok((values, validity))) => {
+                Some(Ok(finish(&self.data_type, values, validity)))
+            }
+            utils::MaybeNext::Some(Err(e)) => Some(Err(e)),
+            utils::MaybeNext::None => None,
+            utils::MaybeNext::More => self.next(),
+        }
+    }
+}
diff --git a/src/io/parquet/read/deserialize/primitive/mod.rs b/src/io/parquet/read/deserialize/primitive/mod.rs
index 70d743dcc2..27d9c27c31 100644
--- a/src/io/parquet/read/deserialize/primitive/mod.rs
+++ b/src/io/parquet/read/deserialize/primitive/mod.rs
@@ -1,7 +1,9 @@
 mod basic;
 mod dictionary;
+mod integer;
 mod nested;
 
 pub use basic::Iter;
 pub use dictionary::{DictIter, NestedDictIter};
+pub use integer::IntegerIter;
 pub use nested::NestedIter;
diff --git a/src/io/parquet/read/deserialize/primitive/nested.rs b/src/io/parquet/read/deserialize/primitive/nested.rs
index 438ca11afa..4798409acb 100644
--- a/src/io/parquet/read/deserialize/primitive/nested.rs
+++ b/src/io/parquet/read/deserialize/primitive/nested.rs
@@ -113,7 +113,7 @@ where
         )
     }
 
-    fn push_valid(&self, state: &mut Self::State, decoded: &mut Self::DecodedState) {
+    fn push_valid(&self, state: &mut Self::State, decoded: &mut Self::DecodedState) -> Result<()> {
         let (values, validity) = decoded;
         match state {
             State::Optional(page_values) => {
@@ -128,19 +128,24 @@ where
                 values.push(value.unwrap_or_default());
             }
             State::RequiredDictionary(page) => {
-                let op1 = |index: u32| page.dict[index as usize];
-                let value = page.values.next().map(op1);
+                let value = page
+                    .values
+                    .next()
+                    .map(|index| page.dict[index.unwrap() as usize]);
 
                 values.push(value.unwrap_or_default());
             }
             State::OptionalDictionary(page) => {
-                let op1 = |index: u32| page.dict[index as usize];
-                let value = page.values.next().map(op1);
+                let value = page
+                    .values
+                    .next()
+                    .map(|index| page.dict[index.unwrap() as usize]);
 
                 values.push(value.unwrap_or_default());
                 validity.push(true);
             }
         }
+        Ok(())
     }
 
     fn push_null(&self, decoded: &mut Self::DecodedState) {
diff --git a/src/io/parquet/read/deserialize/simple.rs b/src/io/parquet/read/deserialize/simple.rs
index 96a27228e9..28dfaa4812 100644
--- a/src/io/parquet/read/deserialize/simple.rs
+++ b/src/io/parquet/read/deserialize/simple.rs
@@ -71,14 +71,14 @@ pub fn page_iter_to_arrays<'a, I: Pages + 'a>(
     Ok(match data_type.to_logical_type() {
         Null => null::iter_to_arrays(pages, data_type, chunk_size, num_rows),
         Boolean => dyn_iter(boolean::Iter::new(pages, data_type, chunk_size, num_rows)),
-        UInt8 => dyn_iter(iden(primitive::Iter::new(
+        UInt8 => dyn_iter(iden(primitive::IntegerIter::new(
             pages,
             data_type,
             num_rows,
             chunk_size,
             |x: i32| x as u8,
         ))),
-        UInt16 => dyn_iter(iden(primitive::Iter::new(
+        UInt16 => dyn_iter(iden(primitive::IntegerIter::new(
             pages,
             data_type,
             num_rows,
@@ -86,7 +86,7 @@ pub fn page_iter_to_arrays<'a, I: Pages + 'a>(
             |x: i32| x as u16,
         ))),
         UInt32 => match physical_type {
-            PhysicalType::Int32 => dyn_iter(iden(primitive::Iter::new(
+            PhysicalType::Int32 => dyn_iter(iden(primitive::IntegerIter::new(
                 pages,
                 data_type,
                 num_rows,
@@ -94,7 +94,7 @@ pub fn page_iter_to_arrays<'a, I: Pages + 'a>(
                 |x: i32| x as u32,
             ))),
             // some implementations of parquet write arrow's u32 into i64.
-            PhysicalType::Int64 => dyn_iter(iden(primitive::Iter::new(
+            PhysicalType::Int64 => dyn_iter(iden(primitive::IntegerIter::new(
                 pages,
                 data_type,
                 num_rows,
@@ -108,21 +108,21 @@ pub fn page_iter_to_arrays<'a, I: Pages + 'a>(
                 )))
             }
         },
-        Int8 => dyn_iter(iden(primitive::Iter::new(
+        Int8 => dyn_iter(iden(primitive::IntegerIter::new(
             pages,
             data_type,
             num_rows,
             chunk_size,
             |x: i32| x as i8,
         ))),
-        Int16 => dyn_iter(iden(primitive::Iter::new(
+        Int16 => dyn_iter(iden(primitive::IntegerIter::new(
             pages,
             data_type,
             num_rows,
             chunk_size,
             |x: i32| x as i16,
         ))),
-        Int32 | Date32 | Time32(_) => dyn_iter(iden(primitive::Iter::new(
+        Int32 | Date32 | Time32(_) => dyn_iter(iden(primitive::IntegerIter::new(
             pages,
             data_type,
             num_rows,
@@ -200,14 +200,14 @@ pub fn page_iter_to_arrays<'a, I: Pages + 'a>(
         }
 
         Decimal(_, _) => match physical_type {
-            PhysicalType::Int32 => dyn_iter(iden(primitive::Iter::new(
+            PhysicalType::Int32 => dyn_iter(iden(primitive::IntegerIter::new(
                 pages,
                 data_type,
                 num_rows,
                 chunk_size,
                 |x: i32| x as i128,
             ))),
-            PhysicalType::Int64 => dyn_iter(iden(primitive::Iter::new(
+            PhysicalType::Int64 => dyn_iter(iden(primitive::IntegerIter::new(
                 pages,
                 data_type,
                 num_rows,
@@ -250,14 +250,14 @@ pub fn page_iter_to_arrays<'a, I: Pages + 'a>(
         },
 
         // INT64
-        Int64 | Date64 | Time64(_) | Duration(_) => dyn_iter(iden(primitive::Iter::new(
+        Int64 | Date64 | Time64(_) | Duration(_) => dyn_iter(iden(primitive::IntegerIter::new(
             pages,
             data_type,
             num_rows,
             chunk_size,
             |x: i64| x as i64,
         ))),
-        UInt64 => dyn_iter(iden(primitive::Iter::new(
+        UInt64 => dyn_iter(iden(primitive::IntegerIter::new(
             pages,
             data_type,
             num_rows,
@@ -368,7 +368,7 @@ fn timestamp<'a, I: Pages + 'a>(
         ));
     }
 
-    let iter = primitive::Iter::new(pages, data_type, num_rows, chunk_size, |x: i64| x);
+    let iter = primitive::IntegerIter::new(pages, data_type, num_rows, chunk_size, |x: i64| x);
     let (factor, is_multiplier) = unifiy_timestmap_unit(logical_type, time_unit);
     match (factor, is_multiplier) {
         (1, _) => Ok(dyn_iter(iden(iter))),
diff --git a/src/io/parquet/read/deserialize/utils.rs b/src/io/parquet/read/deserialize/utils.rs
index b847709e7b..2c137119b4 100644
--- a/src/io/parquet/read/deserialize/utils.rs
+++ b/src/io/parquet/read/deserialize/utils.rs
@@ -126,7 +126,7 @@ impl<'a> PageValidity<'a> for FilteredOptionalPageValidity<'a> {
             (run, offset)
         } else {
             // a new run
-            let run = self.iter.next()?; // no run -> None
+            let run = self.iter.next()?.unwrap(); // no run -> None
             self.current = Some((run, 0));
             return self.next_limited(limit);
         };
@@ -234,7 +234,7 @@ impl<'a> OptionalPageValidity<'a> {
             (run, offset)
         } else {
             // a new run
-            let run = self.iter.next()?; // no run -> None
+            let run = self.iter.next()?.unwrap(); // no run -> None
             self.current = Some((run, 0));
             return self.next_limited(limit);
         };
@@ -492,9 +492,6 @@ pub(super) fn dict_indices_decoder(page: &DataPage) -> Result<hybrid_rle::Hybrid
     let bit_width = indices_buffer[0];
     let indices_buffer = &indices_buffer[1..];
 
-    Ok(hybrid_rle::HybridRleDecoder::new(
-        indices_buffer,
-        bit_width as u32,
-        page.num_values(),
-    ))
+    hybrid_rle::HybridRleDecoder::try_new(indices_buffer, bit_width as u32, page.num_values())
+        .map_err(Error::from)
 }
diff --git a/src/io/parquet/write/dictionary.rs b/src/io/parquet/write/dictionary.rs
index f04acc2fc5..2abffd98a4 100644
--- a/src/io/parquet/write/dictionary.rs
+++ b/src/io/parquet/write/dictionary.rs
@@ -48,20 +48,20 @@ fn serialize_keys_values<K: DictionaryKey>(
         let keys = keys
             .zip(validity.iter())
             .filter_map(|(key, is_valid)| is_valid.then(|| key));
-        let num_bits = utils::get_bit_width(keys.clone().max().unwrap_or(0) as u64) as u8;
+        let num_bits = utils::get_bit_width(keys.clone().max().unwrap_or(0) as u64);
 
         let keys = utils::ExactSizedIter::new(keys, array.len() - validity.unset_bits());
 
         // num_bits as a single byte
-        buffer.push(num_bits);
+        buffer.push(num_bits as u8);
 
         // followed by the encoded indices.
         Ok(encode_u32(buffer, keys, num_bits)?)
     } else {
-        let num_bits = utils::get_bit_width(keys.clone().max().unwrap_or(0) as u64) as u8;
+        let num_bits = utils::get_bit_width(keys.clone().max().unwrap_or(0) as u64);
 
         // num_bits as a single byte
-        buffer.push(num_bits);
+        buffer.push(num_bits as u8);
 
         // followed by the encoded indices.
         Ok(encode_u32(buffer, keys, num_bits)?)
@@ -149,8 +149,7 @@ macro_rules! dyn_prim {
     ($from:ty, $to:ty, $array:expr, $options:expr, $type_:expr) => {{
         let values = $array.values().as_any().downcast_ref().unwrap();
 
-        let mut buffer = vec![];
-        primitive_encode_plain::<$from, $to>(values, false, &mut buffer);
+        let buffer = primitive_encode_plain::<$from, $to>(values, false, vec![]);
         let stats = primitive_build_statistics::<$from, $to>(values, $type_.clone());
         let stats = serialize_statistics(&stats);
         (DictPage::new(buffer, values.len(), false), stats)
diff --git a/src/io/parquet/write/mod.rs b/src/io/parquet/write/mod.rs
index 3b8f48e993..0abc6330a8 100644
--- a/src/io/parquet/write/mod.rs
+++ b/src/io/parquet/write/mod.rs
@@ -79,8 +79,14 @@ pub fn to_parquet_schema(schema: &Schema) -> Result<SchemaDescriptor> {
 /// Note that this is whether this implementation supports it, which is a subset of
 /// what the parquet spec allows.
 pub fn can_encode(data_type: &DataType, encoding: Encoding) -> bool {
+    if let (Encoding::DeltaBinaryPacked, DataType::Decimal(p, _)) =
+        (encoding, data_type.to_logical_type())
+    {
+        return *p <= 18;
+    };
+
     matches!(
-        (encoding, data_type),
+        (encoding, data_type.to_logical_type()),
         (Encoding::Plain, _)
             | (
                 Encoding::DeltaLengthByteArray,
@@ -88,6 +94,24 @@ pub fn can_encode(data_type: &DataType, encoding: Encoding) -> bool {
             )
             | (Encoding::RleDictionary, DataType::Dictionary(_, _, _))
             | (Encoding::PlainDictionary, DataType::Dictionary(_, _, _))
+            | (
+                Encoding::DeltaBinaryPacked,
+                DataType::Null
+                    | DataType::UInt8
+                    | DataType::UInt16
+                    | DataType::UInt32
+                    | DataType::UInt64
+                    | DataType::Int8
+                    | DataType::Int16
+                    | DataType::Int32
+                    | DataType::Date32
+                    | DataType::Time32(_)
+                    | DataType::Int64
+                    | DataType::Date64
+                    | DataType::Time64(_)
+                    | DataType::Timestamp(_, _)
+                    | DataType::Duration(_)
+            )
     )
 }
 
@@ -167,58 +191,66 @@ pub fn array_to_page_simple(
             boolean::array_to_page(array.as_any().downcast_ref().unwrap(), options, type_)
         }
         // casts below MUST match the casts done at the metadata (field -> parquet type).
-        DataType::UInt8 => primitive::array_to_page::<u8, i32>(
+        DataType::UInt8 => primitive::array_to_page_integer::<u8, i32>(
             array.as_any().downcast_ref().unwrap(),
             options,
             type_,
+            encoding,
         ),
-        DataType::UInt16 => primitive::array_to_page::<u16, i32>(
+        DataType::UInt16 => primitive::array_to_page_integer::<u16, i32>(
             array.as_any().downcast_ref().unwrap(),
             options,
             type_,
+            encoding,
         ),
-        DataType::UInt32 => primitive::array_to_page::<u32, i32>(
+        DataType::UInt32 => primitive::array_to_page_integer::<u32, i32>(
             array.as_any().downcast_ref().unwrap(),
             options,
             type_,
+            encoding,
         ),
-        DataType::UInt64 => primitive::array_to_page::<u64, i64>(
+        DataType::UInt64 => primitive::array_to_page_integer::<u64, i64>(
             array.as_any().downcast_ref().unwrap(),
             options,
             type_,
+            encoding,
         ),
-        DataType::Int8 => primitive::array_to_page::<i8, i32>(
+        DataType::Int8 => primitive::array_to_page_integer::<i8, i32>(
             array.as_any().downcast_ref().unwrap(),
             options,
             type_,
+            encoding,
         ),
-        DataType::Int16 => primitive::array_to_page::<i16, i32>(
+        DataType::Int16 => primitive::array_to_page_integer::<i16, i32>(
             array.as_any().downcast_ref().unwrap(),
             options,
             type_,
+            encoding,
         ),
         DataType::Int32 | DataType::Date32 | DataType::Time32(_) => {
-            primitive::array_to_page::<i32, i32>(
+            primitive::array_to_page_integer::<i32, i32>(
                 array.as_any().downcast_ref().unwrap(),
                 options,
                 type_,
+                encoding,
             )
         }
         DataType::Int64
         | DataType::Date64
         | DataType::Time64(_)
         | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => primitive::array_to_page::<i64, i64>(
+        | DataType::Duration(_) => primitive::array_to_page_integer::<i64, i64>(
             array.as_any().downcast_ref().unwrap(),
             options,
             type_,
+            encoding,
         ),
-        DataType::Float32 => primitive::array_to_page::<f32, f32>(
+        DataType::Float32 => primitive::array_to_page_plain::<f32, f32>(
             array.as_any().downcast_ref().unwrap(),
             options,
             type_,
         ),
-        DataType::Float64 => primitive::array_to_page::<f64, f64>(
+        DataType::Float64 => primitive::array_to_page_plain::<f64, f64>(
             array.as_any().downcast_ref().unwrap(),
             options,
             type_,
@@ -249,7 +281,7 @@ pub fn array_to_page_simple(
         ),
         DataType::Null => {
             let array = Int32Array::new_null(DataType::Int32, array.len());
-            primitive::array_to_page::<i32, i32>(&array, options, type_)
+            primitive::array_to_page_plain::<i32, i32>(&array, options, type_)
         }
         DataType::Interval(IntervalUnit::YearMonth) => {
             let type_ = type_;
@@ -327,7 +359,7 @@ pub fn array_to_page_simple(
 
                 let array =
                     PrimitiveArray::<i32>::new(DataType::Int32, values, array.validity().cloned());
-                primitive::array_to_page::<i32, i32>(&array, options, type_)
+                primitive::array_to_page_integer::<i32, i32>(&array, options, type_, encoding)
             } else if precision <= 18 {
                 let values = array
                     .values()
@@ -338,7 +370,7 @@ pub fn array_to_page_simple(
 
                 let array =
                     PrimitiveArray::<i64>::new(DataType::Int64, values, array.validity().cloned());
-                primitive::array_to_page::<i64, i64>(&array, options, type_)
+                primitive::array_to_page_integer::<i64, i64>(&array, options, type_, encoding)
             } else {
                 let size = decimal_length_from_precision(precision);
 
diff --git a/src/io/parquet/write/nested/mod.rs b/src/io/parquet/write/nested/mod.rs
index 39a3f4eddf..9fd603a524 100644
--- a/src/io/parquet/write/nested/mod.rs
+++ b/src/io/parquet/write/nested/mod.rs
@@ -33,7 +33,7 @@ fn write_rep_levels(buffer: &mut Vec<u8>, nested: &[Nested], version: Version) -
     if max_level == 0 {
         return Ok(());
     }
-    let num_bits = get_bit_width(max_level) as u8;
+    let num_bits = get_bit_width(max_level);
 
     let levels = rep::RepLevelsIter::new(nested);
 
@@ -61,13 +61,10 @@ fn write_def_levels(buffer: &mut Vec<u8>, nested: &[Nested], version: Version) -
     if max_level == 0 {
         return Ok(());
     }
-    let num_bits = get_bit_width(max_level) as u8;
+    let num_bits = get_bit_width(max_level);
 
     let levels = def::DefLevelsIter::new(nested);
 
-    let mut buffer1 = vec![];
-    encode_u32(&mut buffer1, def::DefLevelsIter::new(nested), num_bits).unwrap();
-
     match version {
         Version::V1 => write_levels_v1(buffer, move |buffer: &mut Vec<u8>| {
             encode_u32(buffer, levels, num_bits)?;
diff --git a/src/io/parquet/write/primitive/basic.rs b/src/io/parquet/write/primitive/basic.rs
index 36f1f20e1c..3da58c54fd 100644
--- a/src/io/parquet/write/primitive/basic.rs
+++ b/src/io/parquet/write/primitive/basic.rs
@@ -1,52 +1,127 @@
 use parquet2::{
+    encoding::delta_bitpacked::encode,
     encoding::Encoding,
     page::DataPage,
     schema::types::PrimitiveType,
     statistics::{serialize_statistics, PrimitiveStatistics},
-    types::NativeType,
+    types::NativeType as ParquetNativeType,
 };
 
 use super::super::utils;
 use super::super::WriteOptions;
 use crate::{
     array::{Array, PrimitiveArray},
-    error::Result,
-    io::parquet::read::schema::is_nullable,
-    types::NativeType as ArrowNativeType,
+    error::Error,
+    io::parquet::{read::schema::is_nullable, write::utils::ExactSizedIter},
+    types::NativeType,
 };
 
-pub(crate) fn encode_plain<T, R>(array: &PrimitiveArray<T>, is_optional: bool, buffer: &mut Vec<u8>)
+pub(crate) fn encode_plain<T, P>(
+    array: &PrimitiveArray<T>,
+    is_optional: bool,
+    mut buffer: Vec<u8>,
+) -> Vec<u8>
 where
-    T: ArrowNativeType,
-    R: NativeType,
-    T: num_traits::AsPrimitive<R>,
+    T: NativeType,
+    P: ParquetNativeType,
+    T: num_traits::AsPrimitive<P>,
 {
     if is_optional {
+        buffer.reserve(std::mem::size_of::<P>() * (array.len() - array.null_count()));
         // append the non-null values
         array.iter().for_each(|x| {
             if let Some(x) = x {
-                let parquet_native: R = x.as_();
+                let parquet_native: P = x.as_();
                 buffer.extend_from_slice(parquet_native.to_le_bytes().as_ref())
             }
         });
     } else {
+        buffer.reserve(std::mem::size_of::<P>() * array.len());
         // append all values
         array.values().iter().for_each(|x| {
-            let parquet_native: R = x.as_();
+            let parquet_native: P = x.as_();
             buffer.extend_from_slice(parquet_native.to_le_bytes().as_ref())
         });
     }
+    buffer
+}
+
+pub(crate) fn encode_delta<T, P>(
+    array: &PrimitiveArray<T>,
+    is_optional: bool,
+    mut buffer: Vec<u8>,
+) -> Vec<u8>
+where
+    T: NativeType,
+    P: ParquetNativeType,
+    T: num_traits::AsPrimitive<P>,
+    P: num_traits::AsPrimitive<i64>,
+{
+    if is_optional {
+        // append the non-null values
+        let iterator = array.iter().flatten().map(|x| {
+            let parquet_native: P = x.as_();
+            let integer: i64 = parquet_native.as_();
+            integer
+        });
+        let iterator = ExactSizedIter::new(iterator, array.len() - array.null_count());
+        encode(iterator, &mut buffer)
+    } else {
+        // append all values
+        let iterator = array.values().iter().map(|x| {
+            let parquet_native: P = x.as_();
+            let integer: i64 = parquet_native.as_();
+            integer
+        });
+        encode(iterator, &mut buffer)
+    }
+    buffer
+}
+
+pub fn array_to_page_plain<T, P>(
+    array: &PrimitiveArray<T>,
+    options: WriteOptions,
+    type_: PrimitiveType,
+) -> Result<DataPage, Error>
+where
+    T: NativeType,
+    P: ParquetNativeType,
+    T: num_traits::AsPrimitive<P>,
+{
+    array_to_page(array, options, type_, Encoding::Plain, encode_plain)
+}
+
+pub fn array_to_page_integer<T, P>(
+    array: &PrimitiveArray<T>,
+    options: WriteOptions,
+    type_: PrimitiveType,
+    encoding: Encoding,
+) -> Result<DataPage, Error>
+where
+    T: NativeType,
+    P: ParquetNativeType,
+    T: num_traits::AsPrimitive<P>,
+    P: num_traits::AsPrimitive<i64>,
+{
+    match encoding {
+        Encoding::DeltaBinaryPacked => array_to_page(array, options, type_, encoding, encode_delta),
+        Encoding::Plain => array_to_page(array, options, type_, encoding, encode_plain),
+        other => Err(Error::nyi(format!("Encoding integer as {other:?}"))),
+    }
 }
 
-pub fn array_to_page<T, R>(
+pub fn array_to_page<T, P, F: Fn(&PrimitiveArray<T>, bool, Vec<u8>) -> Vec<u8>>(
     array: &PrimitiveArray<T>,
     options: WriteOptions,
     type_: PrimitiveType,
-) -> Result<DataPage>
+    encoding: Encoding,
+    encode: F,
+) -> Result<DataPage, Error>
 where
-    T: ArrowNativeType,
-    R: NativeType,
-    T: num_traits::AsPrimitive<R>,
+    T: NativeType,
+    P: ParquetNativeType,
+    // constraint required to build statistics
+    T: num_traits::AsPrimitive<P>,
 {
     let is_optional = is_nullable(&type_.field_info);
 
@@ -63,7 +138,7 @@ where
 
     let definition_levels_byte_length = buffer.len();
 
-    encode_plain(array, is_optional, &mut buffer);
+    let buffer = encode(array, is_optional, buffer);
 
     let statistics = if options.write_statistics {
         Some(serialize_statistics(&build_statistics(
@@ -84,20 +159,20 @@ where
         statistics,
         type_,
         options,
-        Encoding::Plain,
+        encoding,
     )
 }
 
-pub fn build_statistics<T, R>(
+pub fn build_statistics<T, P>(
     array: &PrimitiveArray<T>,
     primitive_type: PrimitiveType,
-) -> PrimitiveStatistics<R>
+) -> PrimitiveStatistics<P>
 where
-    T: ArrowNativeType,
-    R: NativeType,
-    T: num_traits::AsPrimitive<R>,
+    T: NativeType,
+    P: ParquetNativeType,
+    T: num_traits::AsPrimitive<P>,
 {
-    PrimitiveStatistics::<R> {
+    PrimitiveStatistics::<P> {
         primitive_type,
         null_count: Some(array.null_count() as i64),
         distinct_count: None,
@@ -105,7 +180,7 @@ where
             .iter()
             .flatten()
             .map(|x| {
-                let x: R = x.as_();
+                let x: P = x.as_();
                 x
             })
             .max_by(|x, y| x.ord(y)),
@@ -113,7 +188,7 @@ where
             .iter()
             .flatten()
             .map(|x| {
-                let x: R = x.as_();
+                let x: P = x.as_();
                 x
             })
             .min_by(|x, y| x.ord(y)),
diff --git a/src/io/parquet/write/primitive/mod.rs b/src/io/parquet/write/primitive/mod.rs
index eec1d695d1..7bbdc21943 100644
--- a/src/io/parquet/write/primitive/mod.rs
+++ b/src/io/parquet/write/primitive/mod.rs
@@ -1,7 +1,8 @@
 mod basic;
 mod nested;
 
-pub use basic::array_to_page;
+pub use basic::array_to_page_integer;
+pub use basic::array_to_page_plain;
 pub(crate) use basic::build_statistics;
 pub(crate) use basic::encode_plain;
 pub use nested::array_to_page as nested_array_to_page;
diff --git a/src/io/parquet/write/primitive/nested.rs b/src/io/parquet/write/primitive/nested.rs
index ffbfde6554..e2909b02ce 100644
--- a/src/io/parquet/write/primitive/nested.rs
+++ b/src/io/parquet/write/primitive/nested.rs
@@ -31,7 +31,7 @@ where
     let (repetition_levels_byte_length, definition_levels_byte_length) =
         nested::write_rep_and_def(options.version, nested, &mut buffer)?;
 
-    encode_plain(array, is_optional, &mut buffer);
+    let buffer = encode_plain(array, is_optional, buffer);
 
     let statistics = if options.write_statistics {
         Some(serialize_statistics(&build_statistics(
diff --git a/src/io/parquet/write/row_group.rs b/src/io/parquet/write/row_group.rs
index 5558bab49e..00fe53278d 100644
--- a/src/io/parquet/write/row_group.rs
+++ b/src/io/parquet/write/row_group.rs
@@ -45,7 +45,7 @@ pub fn row_group_iter<A: AsRef<dyn Array> + 'static + Send + Sync>(
                         let pages = DynIter::new(
                             pages
                                 .into_iter()
-                                .map(|x| x.map_err(|e| ParquetError::General(e.to_string()))),
+                                .map(|x| x.map_err(|e| ParquetError::OutOfSpec(e.to_string()))),
                         );
 
                         let compressed_pages = Compressor::new(pages, options.compression, vec![])


Test patch:
diff --git a/arrow-parquet-integration-testing/main.py b/arrow-parquet-integration-testing/main.py
index 42e19e81b8..a880af617d 100644
--- a/arrow-parquet-integration-testing/main.py
+++ b/arrow-parquet-integration-testing/main.py
@@ -10,7 +10,12 @@ def get_file_path(file: str):
 
 
 def _prepare(
-    file: str, version: str, compression: str, encoding_utf8: str, projection=None
+    file: str,
+    version: str,
+    compression: str,
+    encoding_utf8: str,
+    encoding_int: str,
+    projection=None,
 ):
     write = f"{file}.parquet"
 
@@ -26,6 +31,8 @@ def _prepare(
         version,
         "--encoding-utf8",
         encoding_utf8,
+        "--encoding-int",
+        encoding_int,
         "--compression",
         compression,
     ]
@@ -38,7 +45,7 @@ def _prepare(
     return write
 
 
-def _expected(file: str):
+def _expected(file: str) -> pyarrow.Table:
     return pyarrow.ipc.RecordBatchFileReader(get_file_path(file)).read_all()
 
 
@@ -75,16 +82,19 @@ def variations():
             # "generated_custom_metadata",
         ]:
             # pyarrow does not support decoding "delta"-encoded values.
-            # for encoding in ["plain", "delta"]:
-            for encoding in ["plain"]:
+            for encoding_int in ["plain", "delta"]:
+                if encoding_int == "delta" and file in {"generated_primitive", "generated_null"}:
+                    # see https://issues.apache.org/jira/browse/ARROW-17465
+                    continue
+
                 for compression in ["uncompressed", "zstd", "snappy"]:
-                    yield (version, file, compression, encoding)
+                    yield (version, file, compression, "plain", encoding_int)
 
 
 if __name__ == "__main__":
-    for (version, file, compression, encoding_utf8) in variations():
+    for (version, file, compression, encoding_utf8, encoding_int) in variations():
         expected = _expected(file)
-        path = _prepare(file, version, compression, encoding_utf8)
+        path = _prepare(file, version, compression, encoding_utf8, encoding_int)
 
         table = pq.read_table(path)
         os.remove(path)
@@ -95,4 +105,4 @@ if __name__ == "__main__":
             if str(c1.type) in ["month_interval", "day_time_interval"]:
                 # pyarrow does not support interval types from parquet
                 continue
-            assert c1 == c2
+            assert c1 == c2, (c1, c2)
diff --git a/arrow-parquet-integration-testing/main_spark.py b/arrow-parquet-integration-testing/main_spark.py
index c33a1aa63a..e29655fb46 100644
--- a/arrow-parquet-integration-testing/main_spark.py
+++ b/arrow-parquet-integration-testing/main_spark.py
@@ -7,7 +7,13 @@ import pyspark.sql
 from main import _prepare, _expected
 
 
-def test(file: str, version: str, column, compression: str, encoding: str):
+def test(
+    file: str,
+    version: str,
+    column: str,
+    compression: str,
+    encoding: str,
+):
     """
     Tests that pyspark can read a parquet file written by arrow2.
 
@@ -16,13 +22,13 @@ def test(file: str, version: str, column, compression: str, encoding: str):
     In pyspark: read (written) parquet to Python
     assert that they are equal
     """
-    # write parquet
-    path = _prepare(file, version, compression, encoding, [column[1]])
-
     # read IPC to Python
     expected = _expected(file)
-    expected = next(c for i, c in enumerate(expected) if i == column[1])
-    expected = expected.combine_chunks().tolist()
+    column_index = next(i for i, c in enumerate(expected.column_names) if c == column)
+    expected = expected[column].combine_chunks().tolist()
+
+    # write parquet
+    path = _prepare(file, version, compression, encoding, encoding, [column_index])
 
     # read parquet to Python
     spark = pyspark.sql.SparkSession.builder.config(
@@ -31,28 +37,34 @@ def test(file: str, version: str, column, compression: str, encoding: str):
         "false",
     ).getOrCreate()
 
-    result = spark.read.parquet(path).select(column[0]).collect()
-    result = [r[column[0]] for r in result]
+    result = spark.read.parquet(path).select(column).collect()
+    result = [r[column] for r in result]
     os.remove(path)
 
     # assert equality
     assert expected == result
 
 
-test("generated_primitive", "2", ("utf8_nullable", 24), "uncompressed", "delta")
-test("generated_primitive", "2", ("utf8_nullable", 24), "snappy", "delta")
+test("generated_null", "2", "f1", "uncompressed", "delta")
+
+test("generated_primitive", "2", "utf8_nullable", "uncompressed", "delta")
+test("generated_primitive", "2", "utf8_nullable", "snappy", "delta")
+test("generated_primitive", "2", "int32_nullable", "uncompressed", "delta")
+test("generated_primitive", "2", "int32_nullable", "snappy", "delta")
+test("generated_primitive", "2", "int16_nullable", "uncompressed", "delta")
+test("generated_primitive", "2", "int16_nullable", "snappy", "delta")
 
-test("generated_dictionary", "1", ("dict0", 0), "uncompressed", "plain")
-test("generated_dictionary", "1", ("dict0", 0), "snappy", "plain")
-test("generated_dictionary", "2", ("dict0", 0), "uncompressed", "plain")
-test("generated_dictionary", "2", ("dict0", 0), "snappy", "plain")
+test("generated_dictionary", "1", "dict0", "uncompressed", "plain")
+test("generated_dictionary", "1", "dict0", "snappy", "plain")
+test("generated_dictionary", "2", "dict0", "uncompressed", "plain")
+test("generated_dictionary", "2", "dict0", "snappy", "plain")
 
-test("generated_dictionary", "1", ("dict1", 1), "uncompressed", "plain")
-test("generated_dictionary", "1", ("dict1", 1), "snappy", "plain")
-test("generated_dictionary", "2", ("dict1", 1), "uncompressed", "plain")
-test("generated_dictionary", "2", ("dict1", 1), "snappy", "plain")
+test("generated_dictionary", "1", "dict1", "uncompressed", "plain")
+test("generated_dictionary", "1", "dict1", "snappy", "plain")
+test("generated_dictionary", "2", "dict1", "uncompressed", "plain")
+test("generated_dictionary", "2", "dict1", "snappy", "plain")
 
-test("generated_dictionary", "1", ("dict2", 2), "uncompressed", "plain")
-test("generated_dictionary", "1", ("dict2", 2), "snappy", "plain")
-test("generated_dictionary", "2", ("dict2", 2), "uncompressed", "plain")
-test("generated_dictionary", "2", ("dict2", 2), "snappy", "plain")
+test("generated_dictionary", "1", "dict2", "uncompressed", "plain")
+test("generated_dictionary", "1", "dict2", "snappy", "plain")
+test("generated_dictionary", "2", "dict2", "uncompressed", "plain")
+test("generated_dictionary", "2", "dict2", "snappy", "plain")
diff --git a/arrow-parquet-integration-testing/src/main.rs b/arrow-parquet-integration-testing/src/main.rs
index d79da94318..831dd6b5f0 100644
--- a/arrow-parquet-integration-testing/src/main.rs
+++ b/arrow-parquet-integration-testing/src/main.rs
@@ -1,10 +1,9 @@
 use std::fs::File;
-use std::{io::Read};
+use std::io::Read;
 
 use arrow2::array::Array;
 use arrow2::io::ipc::IpcField;
 use arrow2::{
-    AHashMap,
     chunk::Chunk,
     datatypes::{DataType, Schema},
     error::Result,
@@ -16,6 +15,7 @@ use arrow2::{
             RowGroupIterator, Version as ParquetVersion, WriteOptions,
         },
     },
+    AHashMap,
 };
 use clap::Parser;
 use flate2::read::GzDecoder;
@@ -110,6 +110,8 @@ struct Args {
     projection: Option<String>,
     #[clap(short, long, arg_enum, help = "encoding scheme for utf8", default_value_t = EncodingScheme::Plain)]
     encoding_utf8: EncodingScheme,
+    #[clap(short('i'), long, arg_enum, help = "encoding scheme for int", default_value_t = EncodingScheme::Plain)]
+    encoding_int: EncodingScheme,
     #[clap(short, long, arg_enum)]
     compression: Compression,
 }
@@ -178,6 +180,13 @@ fn main() -> Result<()> {
         .map(|f| {
             transverse(&f.data_type, |dt| match dt {
                 DataType::Dictionary(..) => Encoding::RleDictionary,
+                DataType::Int32 => {
+                    if args.encoding_int == EncodingScheme::Delta {
+                        Encoding::DeltaBinaryPacked
+                    } else {
+                        Encoding::Plain
+                    }
+                }
                 DataType::Utf8 | DataType::LargeUtf8 => {
                     if args.encoding_utf8 == EncodingScheme::Delta {
                         Encoding::DeltaLengthByteArray
diff --git a/tests/it/io/parquet/read_indexes.rs b/tests/it/io/parquet/read_indexes.rs
index bff9346006..bfa5f2a3c3 100644
--- a/tests/it/io/parquet/read_indexes.rs
+++ b/tests/it/io/parquet/read_indexes.rs
@@ -197,6 +197,30 @@ fn indexed_optional_i32() -> Result<()> {
     read_with_indexes(pages(&[&array21, &array22], Encoding::Plain)?, expected)
 }
 
+#[test]
+fn indexed_optional_i32_delta() -> Result<()> {
+    let array21 = Int32Array::from([Some(1), Some(2), None]);
+    let array22 = Int32Array::from([None, Some(5), Some(6)]);
+    let expected = Int32Array::from_slice([5]).boxed();
+
+    read_with_indexes(
+        pages(&[&array21, &array22], Encoding::DeltaBinaryPacked)?,
+        expected,
+    )
+}
+
+#[test]
+fn indexed_required_i32_delta() -> Result<()> {
+    let array21 = Int32Array::from_slice([1, 2, 3]);
+    let array22 = Int32Array::from_slice([4, 5, 6]);
+    let expected = Int32Array::from_slice([5]).boxed();
+
+    read_with_indexes(
+        pages(&[&array21, &array22], Encoding::DeltaBinaryPacked)?,
+        expected,
+    )
+}
+
 #[test]
 fn indexed_optional_utf8() -> Result<()> {
     let array21 = Utf8Array::<i32>::from([Some("a"), Some("b"), None]);
diff --git a/tests/it/io/parquet/write.rs b/tests/it/io/parquet/write.rs
index f95b53331c..dfeed7ccd0 100644
--- a/tests/it/io/parquet/write.rs
+++ b/tests/it/io/parquet/write.rs
@@ -92,6 +92,28 @@ fn int64_optional_v2() -> Result<()> {
     )
 }
 
+#[test]
+fn int64_optional_delta() -> Result<()> {
+    round_trip(
+        "int64",
+        "nullable",
+        Version::V2,
+        CompressionOptions::Uncompressed,
+        vec![Encoding::DeltaBinaryPacked],
+    )
+}
+
+#[test]
+fn int64_required_delta() -> Result<()> {
+    round_trip(
+        "int64",
+        "required",
+        Version::V2,
+        CompressionOptions::Uncompressed,
+        vec![Encoding::DeltaBinaryPacked],
+    )
+}
+
 #[cfg(feature = "io_parquet_compression")]
 #[test]
 fn int64_optional_v2_compressed() -> Result<()> {