Repository: jorgecarleitao/arrow2
Pull request #208 (2021-07-19 21:53:08)

Added support to read and write delta-encoded parquet.
This PR adds support to read and write [Delta-length encoded](https://github.com/apache/parquet-format/blob/master/Encodings.md#delta-length-byte-array-delta_length_byte_array--6) utf8 and binary to and from parquet. This encoding lends well to both arrow and compression, as values are encoded back to back.

Reading and writing with this encoding results in a higher compression at no performance difference. The major downside is that pyarrow does not support it and spark seems to require a config option to support it 🙄

```
write utf8 2^20         time:   [20.590 ms 20.744 ms 20.920 ms]
write utf8 delta 2^20   time:   [19.031 ms 19.108 ms 19.194 ms]
```

This PR includes:
* an integration test against pyspark (reading from arrow2)
* a round-trip across arrow2

integration tests against pyarrow are skipped since pyarrow does not support this encoding yet. Filed [ARROW-13388]( https://issues.apache.org/jira/browse/ARROW-13388) to track it.

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

[![Impacted file tree graph](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/graphs/tree.svg?width=650&height=150&src=pr&token=AgyTF60R3D&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)

```diff
@@            Coverage Diff             @@
##             main     #208      +/-   ##
==========================================
- Coverage   76.85%   76.81%   -0.04%     
==========================================
  Files         226      226              
  Lines       19298    19447     +149     
==========================================
+ Hits        14832    14939     +107     
- Misses       4466     4508      +42     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/array/binary/iterator.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2JpbmFyeS9pdGVyYXRvci5ycw==) | `86.66% <ø> (ø)` | |
| [src/array/utf8/iterator.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L3V0ZjgvaXRlcmF0b3IucnM=) | `66.66% <ø> (ø)` | |
| [src/bitmap/utils/iterator.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2JpdG1hcC91dGlscy9pdGVyYXRvci5ycw==) | `100.00% <ø> (ø)` | |
| [src/bitmap/utils/zip\_validity.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2JpdG1hcC91dGlscy96aXBfdmFsaWRpdHkucnM=) | `94.18% <0.00%> (-4.60%)` | :arrow_down: |
| [src/io/parquet/read/binary/nested.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvcmVhZC9iaW5hcnkvbmVzdGVkLnJz) | `87.80% <0.00%> (ø)` | |
| [src/io/parquet/write/record\_batch.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvcmVjb3JkX2JhdGNoLnJz) | `0.00% <0.00%> (ø)` | |
| [src/array/specification.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L3NwZWNpZmljYXRpb24ucnM=) | `69.76% <50.00%> (-4.52%)` | :arrow_down: |
| [src/io/parquet/write/binary/basic.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvYmluYXJ5L2Jhc2ljLnJz) | `85.13% <59.09%> (-11.10%)` | :arrow_down: |
| [src/io/parquet/write/utils.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvdXRpbHMucnM=) | `83.87% <60.00%> (-4.60%)` | :arrow_down: |
| [src/io/parquet/write/utf8/basic.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvdXRmOC9iYXNpYy5ycw==) | `92.00% <66.66%> (-3.56%)` | :arrow_down: |
| ... and [11 more](https://codecov.io/gh/jorgecarleitao/arrow2/pull/208/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | |

------

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


Patch:
diff --git a/.github/workflows/integration-parquet.yml b/.github/workflows/integration-parquet.yml
index 3ef470d4b7..23d2a27eaf 100644
--- a/.github/workflows/integration-parquet.yml
+++ b/.github/workflows/integration-parquet.yml
@@ -46,5 +46,7 @@ jobs:
           python -m venv venv
           source venv/bin/activate
           pip install --upgrade pip
-          pip install pyarrow
+          pip install pyarrow pyspark
           python main.py
+          # test delta encoding against spark (pyarrow does not support it)
+          python main_spark.py
diff --git a/benches/write_parquet.rs b/benches/write_parquet.rs
index d26a6cb56e..df3340a0d4 100644
--- a/benches/write_parquet.rs
+++ b/benches/write_parquet.rs
@@ -8,7 +8,7 @@ use arrow2::error::Result;
 use arrow2::io::parquet::write::*;
 use arrow2::util::bench_util::{create_boolean_array, create_primitive_array, create_string_array};
 
-fn write(array: &dyn Array) -> Result<()> {
+fn write(array: &dyn Array, encoding: Encoding) -> Result<()> {
     let field = Field::new("c1", array.data_type().clone(), true);
     let schema = Schema::new(vec![field]);
 
@@ -25,6 +25,7 @@ fn write(array: &dyn Array) -> Result<()> {
             array,
             parquet_schema.columns()[0].clone(),
             options,
+            encoding,
         )),
     ))))));
 
@@ -44,19 +45,27 @@ fn add_benchmark(c: &mut Criterion) {
     (0..=10).step_by(2).for_each(|i| {
         let array = &create_primitive_array::<i64>(1024 * 2usize.pow(i), DataType::Int64, 0.1);
         let a = format!("write i64 2^{}", 10 + i);
-        c.bench_function(&a, |b| b.iter(|| write(array).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| write(array, Encoding::Plain).unwrap()));
     });
 
     (0..=10).step_by(2).for_each(|i| {
         let array = &create_boolean_array(1024 * 2usize.pow(i), 0.1, 0.5);
         let a = format!("write bool 2^{}", 10 + i);
-        c.bench_function(&a, |b| b.iter(|| write(array).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| write(array, Encoding::Plain).unwrap()));
     });
 
     (0..=10).step_by(2).for_each(|i| {
         let array = &create_string_array::<i32>(1024 * 2usize.pow(i), 0.1);
         let a = format!("write utf8 2^{}", 10 + i);
-        c.bench_function(&a, |b| b.iter(|| write(array).unwrap()));
+        c.bench_function(&a, |b| b.iter(|| write(array, Encoding::Plain).unwrap()));
+    });
+
+    (0..=10).step_by(2).for_each(|i| {
+        let array = &create_string_array::<i32>(1024 * 2usize.pow(i), 0.1);
+        let a = format!("write utf8 delta 2^{}", 10 + i);
+        c.bench_function(&a, |b| {
+            b.iter(|| write(array, Encoding::DeltaLengthByteArray).unwrap())
+        });
     });
 }
 
diff --git a/examples/parquet_write.rs b/examples/parquet_write.rs
index 70b374bf35..52e70f417d 100644
--- a/examples/parquet_write.rs
+++ b/examples/parquet_write.rs
@@ -7,7 +7,7 @@ use arrow2::{
     datatypes::{Field, Schema},
     error::Result,
     io::parquet::write::{
-        array_to_page, write_file, CompressionCodec, DynIter, Version, WriteOptions,
+        array_to_page, write_file, CompressionCodec, DynIter, Encoding, Version, WriteOptions,
     },
 };
 
@@ -19,6 +19,7 @@ fn write_single_array(path: &str, array: &dyn Array, field: Field) -> Result<()>
         compression: CompressionCodec::Uncompressed,
         version: Version::V2,
     };
+    let encoding = Encoding::Plain;
 
     // map arrow fields to parquet fields
     let parquet_schema = to_parquet_schema(&schema)?;
@@ -32,7 +33,7 @@ fn write_single_array(path: &str, array: &dyn Array, field: Field) -> Result<()>
     let row_groups = once(Result::Ok(DynIter::new(once(Ok(DynIter::new(
         once(array)
             .zip(parquet_schema.columns().to_vec().into_iter())
-            .map(|(array, descriptor)| array_to_page(array, descriptor, options)),
+            .map(|(array, descriptor)| array_to_page(array, descriptor, options, encoding)),
     ))))));
 
     // Create a new empty file
diff --git a/examples/parquet_write_record.rs b/examples/parquet_write_record.rs
index b5314ac9ca..9b4deb6879 100644
--- a/examples/parquet_write_record.rs
+++ b/examples/parquet_write_record.rs
@@ -8,6 +8,7 @@ use arrow2::{
     io::parquet::write::{write_file, CompressionCodec, RowGroupIterator, Version, WriteOptions},
     record_batch::RecordBatch,
 };
+use parquet2::schema::Encoding;
 
 fn write_batch(path: &str, batch: RecordBatch) -> Result<()> {
     let schema = batch.schema().clone();
@@ -20,7 +21,8 @@ fn write_batch(path: &str, batch: RecordBatch) -> Result<()> {
 
     let iter = vec![Ok(batch)];
 
-    let row_groups = RowGroupIterator::try_new(iter.into_iter(), &schema, options)?;
+    let row_groups =
+        RowGroupIterator::try_new(iter.into_iter(), &schema, options, vec![Encoding::Plain])?;
 
     // Create a new empty file
     let mut file = File::create(path)?;
diff --git a/src/array/binary/iterator.rs b/src/array/binary/iterator.rs
index 5361aae7a2..d0b116f17f 100644
--- a/src/array/binary/iterator.rs
+++ b/src/array/binary/iterator.rs
@@ -6,6 +6,7 @@ use crate::{
 
 use super::BinaryArray;
 
+#[derive(Debug, Clone)]
 pub struct BinaryValueIter<'a, O: Offset> {
     array: &'a BinaryArray<O>,
     index: usize,
diff --git a/src/array/specification.rs b/src/array/specification.rs
index de3c9d6692..5acfc9e1b5 100644
--- a/src/array/specification.rs
+++ b/src/array/specification.rs
@@ -21,7 +21,11 @@ pub unsafe trait Offset:
 {
     fn is_large() -> bool;
 
+    fn to_isize(&self) -> isize;
+
     fn from_usize(value: usize) -> Option<Self>;
+
+    fn from_isize(value: isize) -> Option<Self>;
 }
 
 unsafe impl Offset for i32 {
@@ -34,6 +38,16 @@ unsafe impl Offset for i32 {
     fn from_usize(value: usize) -> Option<Self> {
         Self::try_from(value).ok()
     }
+
+    #[inline]
+    fn from_isize(value: isize) -> Option<Self> {
+        Self::try_from(value).ok()
+    }
+
+    #[inline]
+    fn to_isize(&self) -> isize {
+        *self as isize
+    }
 }
 
 unsafe impl Offset for i64 {
@@ -46,6 +60,16 @@ unsafe impl Offset for i64 {
     fn from_usize(value: usize) -> Option<Self> {
         Some(value as i64)
     }
+
+    #[inline]
+    fn from_isize(value: isize) -> Option<Self> {
+        Self::try_from(value).ok()
+    }
+
+    #[inline]
+    fn to_isize(&self) -> isize {
+        *self as isize
+    }
 }
 
 impl Index for i32 {
diff --git a/src/array/utf8/iterator.rs b/src/array/utf8/iterator.rs
index c2e0400735..5fe1b285f3 100644
--- a/src/array/utf8/iterator.rs
+++ b/src/array/utf8/iterator.rs
@@ -7,6 +7,7 @@ use crate::{
 use super::Utf8Array;
 
 /// Iterator of values of an `Utf8Array`.
+#[derive(Debug, Clone)]
 pub struct Utf8ValuesIter<'a, O: Offset> {
     array: &'a Utf8Array<O>,
     index: usize,
diff --git a/src/bitmap/utils/iterator.rs b/src/bitmap/utils/iterator.rs
index f87a3ed167..87678fa8ba 100644
--- a/src/bitmap/utils/iterator.rs
+++ b/src/bitmap/utils/iterator.rs
@@ -4,6 +4,7 @@ use super::get_bit_unchecked;
 
 /// An iterator over bits according to the [LSB](https://en.wikipedia.org/wiki/Bit_numbering#Least_significant_bit),
 /// i.e. the bytes `[4u8, 128u8]` correspond to `[false, false, true, false, ..., true]`.
+#[derive(Debug, Clone)]
 pub struct BitmapIter<'a> {
     bytes: &'a [u8],
     index: usize,
diff --git a/src/bitmap/utils/zip_validity.rs b/src/bitmap/utils/zip_validity.rs
index cbd01364e7..16442bda10 100644
--- a/src/bitmap/utils/zip_validity.rs
+++ b/src/bitmap/utils/zip_validity.rs
@@ -10,6 +10,16 @@ pub struct ZipValidity<'a, T, I: Iterator<Item = T>> {
     has_validity: bool,
 }
 
+impl<'a, T, I: Iterator<Item = T> + Clone> Clone for ZipValidity<'a, T, I> {
+    fn clone(&self) -> Self {
+        Self {
+            values: self.values.clone(),
+            validity_iter: self.validity_iter.clone(),
+            has_validity: self.has_validity,
+        }
+    }
+}
+
 impl<'a, T, I: Iterator<Item = T>> ZipValidity<'a, T, I> {
     #[inline]
     pub fn new(values: I, validity: &'a Option<Bitmap>) -> Self {
diff --git a/src/io/parquet/mod.rs b/src/io/parquet/mod.rs
index 673d288771..def7c45e52 100644
--- a/src/io/parquet/mod.rs
+++ b/src/io/parquet/mod.rs
@@ -429,6 +429,7 @@ mod tests_integration {
                         array.as_ref(),
                         type_,
                         options,
+                        Encoding::Plain,
                     ))))
                 },
             ));
diff --git a/src/io/parquet/read/binary/basic.rs b/src/io/parquet/read/binary/basic.rs
index 600420742c..2ca0ec03e1 100644
--- a/src/io/parquet/read/binary/basic.rs
+++ b/src/io/parquet/read/binary/basic.rs
@@ -1,5 +1,5 @@
 use parquet2::{
-    encoding::{bitpacking, hybrid_rle, uleb128, Encoding},
+    encoding::{bitpacking, delta_length_byte_array, hybrid_rle, uleb128, Encoding},
     metadata::{ColumnChunkMetaData, ColumnDescriptor},
     read::{levels, BinaryPageDict, Page, PageHeader, StreamingIterator},
 };
@@ -83,7 +83,59 @@ fn read_dict_buffer<O: Offset>(
     }
 }
 
-fn read_optional<O: Offset>(
+fn read_delta_optional<O: Offset>(
+    validity_buffer: &[u8],
+    values_buffer: &[u8],
+    length: usize,
+    offsets: &mut MutableBuffer<O>,
+    values: &mut MutableBuffer<u8>,
+    validity: &mut MutableBitmap,
+) {
+    let mut last_offset = *offsets.as_mut_slice().last().unwrap();
+
+    // values_buffer: first 4 bytes are len, remaining is values
+    let mut values_iterator = delta_length_byte_array::Decoder::new(values_buffer);
+
+    let validity_iterator = hybrid_rle::Decoder::new(&validity_buffer, 1);
+
+    // offsets:
+    for run in validity_iterator {
+        match run {
+            hybrid_rle::HybridEncoded::Bitpacked(packed) => {
+                // the pack may contain more items than needed.
+                let remaining = length - (offsets.len() - 1);
+                let len = std::cmp::min(packed.len() * 8, remaining);
+                for is_valid in BitmapIter::new(packed, 0, len) {
+                    if is_valid {
+                        let value = values_iterator.next().unwrap() as isize;
+                        last_offset += O::from_isize(value).unwrap();
+                    }
+                    offsets.push(last_offset);
+                }
+                validity.extend_from_slice(packed, 0, len);
+            }
+            hybrid_rle::HybridEncoded::Rle(value, additional) => {
+                let is_set = value[0] == 1;
+                validity.extend_constant(additional, is_set);
+                if is_set {
+                    (0..additional).for_each(|_| {
+                        let value = values_iterator.next().unwrap() as isize;
+                        last_offset += O::from_isize(value).unwrap();
+                        offsets.push(last_offset);
+                    })
+                } else {
+                    offsets.extend_constant(additional, last_offset)
+                }
+            }
+        }
+    }
+
+    // values:
+    let new_values = values_iterator.into_values();
+    values.extend_from_slice(new_values);
+}
+
+fn read_plain_optional<O: Offset>(
     validity_buffer: &[u8],
     values_buffer: &[u8],
     length: usize,
@@ -132,7 +184,7 @@ fn read_optional<O: Offset>(
     }
 }
 
-pub(super) fn read_required<O: Offset>(
+pub(super) fn read_plain_required<O: Offset>(
     buffer: &[u8],
     _length: usize,
     offsets: &mut MutableBuffer<O>,
@@ -176,7 +228,15 @@ fn extend_from_page<O: Offset>(
                     values,
                     validity,
                 ),
-                (Encoding::Plain, None, true) => read_optional::<O>(
+                (Encoding::DeltaLengthByteArray, None, true) => read_delta_optional::<O>(
+                    validity_buffer,
+                    values_buffer,
+                    page.num_values(),
+                    offsets,
+                    values,
+                    validity,
+                ),
+                (Encoding::Plain, None, true) => read_plain_optional::<O>(
                     validity_buffer,
                     values_buffer,
                     page.num_values(),
@@ -185,7 +245,7 @@ fn extend_from_page<O: Offset>(
                     validity,
                 ),
                 (Encoding::Plain, None, false) => {
-                    read_required::<O>(page.buffer(), page.num_values(), offsets, values)
+                    read_plain_required::<O>(page.buffer(), page.num_values(), offsets, values)
                 }
                 _ => {
                     return Err(utils::not_implemented(
@@ -201,34 +261,37 @@ fn extend_from_page<O: Offset>(
         PageHeader::V2(header) => {
             let def_level_buffer_length = header.definition_levels_byte_length as usize;
 
-            match (&page.encoding(), page.dictionary_page(), is_optional) {
-                (Encoding::PlainDictionary, Some(dict), true) => {
-                    let (_, validity_buffer, values_buffer) =
-                        levels::split_buffer_v2(page.buffer(), 0, def_level_buffer_length);
-                    read_dict_buffer::<O>(
-                        validity_buffer,
-                        values_buffer,
-                        page.num_values() as u32,
-                        dict.as_any().downcast_ref().unwrap(),
-                        offsets,
-                        values,
-                        validity,
-                    )
-                }
-                (Encoding::Plain, None, true) => {
-                    let (_, validity_buffer, values_buffer) =
-                        levels::split_buffer_v2(page.buffer(), 0, def_level_buffer_length);
-                    read_optional::<O>(
-                        validity_buffer,
-                        values_buffer,
-                        page.num_values(),
-                        offsets,
-                        values,
-                        validity,
-                    )
-                }
+            let (_, validity_buffer, values_buffer) =
+                levels::split_buffer_v2(page.buffer(), 0, def_level_buffer_length);
+
+            match (page.encoding(), page.dictionary_page(), is_optional) {
+                (Encoding::PlainDictionary, Some(dict), true) => read_dict_buffer::<O>(
+                    validity_buffer,
+                    values_buffer,
+                    page.num_values() as u32,
+                    dict.as_any().downcast_ref().unwrap(),
+                    offsets,
+                    values,
+                    validity,
+                ),
+                (Encoding::Plain, None, true) => read_plain_optional::<O>(
+                    validity_buffer,
+                    values_buffer,
+                    page.num_values(),
+                    offsets,
+                    values,
+                    validity,
+                ),
+                (Encoding::DeltaLengthByteArray, None, true) => read_delta_optional::<O>(
+                    validity_buffer,
+                    values_buffer,
+                    page.num_values(),
+                    offsets,
+                    values,
+                    validity,
+                ),
                 (Encoding::Plain, None, false) => {
-                    read_required::<O>(page.buffer(), page.num_values(), offsets, values)
+                    read_plain_required::<O>(page.buffer(), page.num_values(), offsets, values)
                 }
                 _ => {
                     return Err(utils::not_implemented(
diff --git a/src/io/parquet/read/binary/nested.rs b/src/io/parquet/read/binary/nested.rs
index 50d212dd14..d3835a521d 100644
--- a/src/io/parquet/read/binary/nested.rs
+++ b/src/io/parquet/read/binary/nested.rs
@@ -11,7 +11,7 @@ use parquet2::{
 
 use super::super::nested_utils::*;
 use super::super::utils;
-use super::basic::read_required;
+use super::basic::read_plain_required;
 use crate::{
     array::{Array, BinaryArray, Offset, Utf8Array},
     bitmap::MutableBitmap,
@@ -85,7 +85,7 @@ fn read<O: Offset>(
                     validity,
                 )
             } else {
-                read_required(values_buffer, additional, offsets, values)
+                read_plain_required(values_buffer, additional, offsets, values)
             }
 
             let def_levels = RLEDecoder::new(
diff --git a/src/io/parquet/write/binary/basic.rs b/src/io/parquet/write/binary/basic.rs
index 804dfe8817..b9da6ce5cf 100644
--- a/src/io/parquet/write/binary/basic.rs
+++ b/src/io/parquet/write/binary/basic.rs
@@ -1,4 +1,5 @@
 use parquet2::{
+    encoding::{delta_bitpacked, Encoding},
     metadata::ColumnDescriptor,
     read::CompressedPage,
     statistics::{serialize_statistics, BinaryStatistics, ParquetStatistics, Statistics},
@@ -8,7 +9,8 @@ use parquet2::{
 use super::super::utils;
 use crate::{
     array::{Array, BinaryArray, Offset},
-    error::Result,
+    bitmap::Bitmap,
+    error::{ArrowError, Result},
     io::parquet::read::is_type_nullable,
 };
 
@@ -41,6 +43,7 @@ pub fn array_to_page<O: Offset>(
     array: &BinaryArray<O>,
     options: WriteOptions,
     descriptor: ColumnDescriptor,
+    encoding: Encoding,
 ) -> Result<CompressedPage> {
     let validity = array.validity();
     let is_optional = is_type_nullable(descriptor.type_());
@@ -56,7 +59,23 @@ pub fn array_to_page<O: Offset>(
 
     let definition_levels_byte_length = buffer.len();
 
-    encode_plain(array, is_optional, &mut buffer);
+    match encoding {
+        Encoding::Plain => encode_plain(array, is_optional, &mut buffer),
+        Encoding::DeltaLengthByteArray => encode_delta(
+            array.values(),
+            array.offsets(),
+            array.validity(),
+            is_optional,
+            &mut buffer,
+        ),
+        _ => {
+            return Err(ArrowError::InvalidArgumentError(format!(
+                "Datatype {:?} cannot be encoded by {:?} encoding",
+                array.data_type(),
+                encoding
+            )))
+        }
+    }
 
     let uncompressed_page_size = buffer.len();
 
@@ -78,6 +97,7 @@ pub fn array_to_page<O: Offset>(
         statistics,
         descriptor,
         options,
+        encoding,
     )
 }
 
@@ -103,6 +123,38 @@ pub(super) fn build_statistics<O: Offset>(
     serialize_statistics(statistics)
 }
 
+pub(crate) fn encode_delta<O: Offset>(
+    values: &[u8],
+    offsets: &[O],
+    validity: &Option<Bitmap>,
+    is_optional: bool,
+    buffer: &mut Vec<u8>,
+) {
+    if is_optional {
+        if let Some(validity) = validity {
+            let lengths = offsets
+                .windows(2)
+                .map(|w| (w[1] - w[0]).to_isize() as i32)
+                .zip(validity.iter())
+                .flat_map(|(x, is_valid)| if is_valid { Some(x) } else { None });
+            let length = offsets.len() - 1 - validity.null_count();
+            let lengths = utils::ExactSizedIter::new(lengths, length);
+
+            delta_bitpacked::encode(lengths, buffer);
+        } else {
+            let lengths = offsets.windows(2).map(|w| (w[1] - w[0]).to_isize() as i32);
+            delta_bitpacked::encode(lengths, buffer);
+        }
+    } else {
+        let lengths = offsets.windows(2).map(|w| (w[1] - w[0]).to_isize() as i32);
+        delta_bitpacked::encode(lengths, buffer);
+    }
+
+    buffer.extend_from_slice(
+        &values[offsets.first().unwrap().to_usize()..offsets.last().unwrap().to_usize()],
+    )
+}
+
 /// Returns the ordering of two binary values. This corresponds to pyarrows' ordering
 /// of statistics.
 pub(crate) fn ord_binary<'a>(a: &'a [u8], b: &'a [u8]) -> std::cmp::Ordering {
diff --git a/src/io/parquet/write/binary/mod.rs b/src/io/parquet/write/binary/mod.rs
index d6ef2d3836..f3c414b319 100644
--- a/src/io/parquet/write/binary/mod.rs
+++ b/src/io/parquet/write/binary/mod.rs
@@ -2,5 +2,5 @@ mod basic;
 mod nested;
 
 pub use basic::array_to_page;
-pub(super) use basic::ord_binary;
+pub(super) use basic::{encode_delta, ord_binary};
 pub use nested::array_to_page as nested_array_to_page;
diff --git a/src/io/parquet/write/binary/nested.rs b/src/io/parquet/write/binary/nested.rs
index 6484eb9322..6cefd9b45f 100644
--- a/src/io/parquet/write/binary/nested.rs
+++ b/src/io/parquet/write/binary/nested.rs
@@ -1,3 +1,4 @@
+use parquet2::schema::Encoding;
 use parquet2::{metadata::ColumnDescriptor, read::CompressedPage, write::WriteOptions};
 
 use super::super::{levels, utils};
@@ -55,5 +56,6 @@ where
         statistics,
         descriptor,
         options,
+        Encoding::Plain,
     )
 }
diff --git a/src/io/parquet/write/boolean/basic.rs b/src/io/parquet/write/boolean/basic.rs
index 69619ef125..ca1fca9663 100644
--- a/src/io/parquet/write/boolean/basic.rs
+++ b/src/io/parquet/write/boolean/basic.rs
@@ -2,6 +2,7 @@ use parquet2::{
     encoding::hybrid_rle::bitpacked_encode,
     metadata::ColumnDescriptor,
     read::CompressedPage,
+    schema::Encoding,
     statistics::{serialize_statistics, BooleanStatistics, ParquetStatistics, Statistics},
     write::WriteOptions,
 };
@@ -80,6 +81,7 @@ pub fn array_to_page(
         statistics,
         descriptor,
         options,
+        Encoding::Plain,
     )
 }
 
diff --git a/src/io/parquet/write/boolean/nested.rs b/src/io/parquet/write/boolean/nested.rs
index b169d4a64a..0bc6170541 100644
--- a/src/io/parquet/write/boolean/nested.rs
+++ b/src/io/parquet/write/boolean/nested.rs
@@ -1,3 +1,4 @@
+use parquet2::schema::Encoding;
 use parquet2::{metadata::ColumnDescriptor, read::CompressedPage, write::WriteOptions};
 
 use super::super::{levels, utils};
@@ -54,5 +55,6 @@ where
         statistics,
         descriptor,
         options,
+        Encoding::Plain,
     )
 }
diff --git a/src/io/parquet/write/fixed_len_bytes.rs b/src/io/parquet/write/fixed_len_bytes.rs
index be1a996943..3e4d011bd1 100644
--- a/src/io/parquet/write/fixed_len_bytes.rs
+++ b/src/io/parquet/write/fixed_len_bytes.rs
@@ -1,5 +1,5 @@
 use parquet2::{
-    compression::create_codec, metadata::ColumnDescriptor, read::CompressedPage,
+    compression::create_codec, metadata::ColumnDescriptor, read::CompressedPage, schema::Encoding,
     write::WriteOptions,
 };
 
@@ -64,5 +64,6 @@ pub fn array_to_page(
         None,
         descriptor,
         options,
+        Encoding::Plain,
     )
 }
diff --git a/src/io/parquet/write/mod.rs b/src/io/parquet/write/mod.rs
index 8703d4132d..deabba2742 100644
--- a/src/io/parquet/write/mod.rs
+++ b/src/io/parquet/write/mod.rs
@@ -25,6 +25,7 @@ pub use parquet2::{
     compression::CompressionCodec,
     read::CompressedPage,
     schema::types::ParquetType,
+    schema::Encoding,
     write::{DynIter, RowGroupIter},
     write::{Version, WriteOptions},
 };
@@ -87,13 +88,35 @@ where
     )?)
 }
 
+/// Checks whether the `data_type` can be encoded as `encoding`.
+/// Note that this is whether this implementation supports it, which is not necessarily
+/// what the parquet spec allows.
+pub fn can_encode(data_type: &DataType, encoding: Encoding) -> bool {
+    matches!(
+        (encoding, data_type),
+        (Encoding::Plain, _)
+            | (
+                Encoding::DeltaLengthByteArray,
+                DataType::Binary | DataType::LargeBinary | DataType::Utf8 | DataType::LargeUtf8,
+            )
+    )
+}
+
 pub fn array_to_page(
     array: &dyn Array,
     descriptor: ColumnDescriptor,
     options: WriteOptions,
+    encoding: Encoding,
 ) -> Result<CompressedPage> {
-    // using plain encoding format
-    match array.data_type() {
+    let data_type = array.data_type();
+    if !can_encode(data_type, encoding) {
+        return Err(ArrowError::InvalidArgumentError(format!(
+            "The datatype {:?} cannot be encoded by {:?}",
+            data_type, encoding
+        )));
+    }
+
+    match data_type {
         DataType::Boolean => {
             boolean::array_to_page(array.as_any().downcast_ref().unwrap(), options, descriptor)
         }
@@ -154,21 +177,29 @@ pub fn array_to_page(
             options,
             descriptor,
         ),
-        DataType::Utf8 => {
-            utf8::array_to_page::<i32>(array.as_any().downcast_ref().unwrap(), options, descriptor)
-        }
-        DataType::LargeUtf8 => {
-            utf8::array_to_page::<i64>(array.as_any().downcast_ref().unwrap(), options, descriptor)
-        }
+        DataType::Utf8 => utf8::array_to_page::<i32>(
+            array.as_any().downcast_ref().unwrap(),
+            options,
+            descriptor,
+            encoding,
+        ),
+        DataType::LargeUtf8 => utf8::array_to_page::<i64>(
+            array.as_any().downcast_ref().unwrap(),
+            options,
+            descriptor,
+            encoding,
+        ),
         DataType::Binary => binary::array_to_page::<i32>(
             array.as_any().downcast_ref().unwrap(),
             options,
             descriptor,
+            encoding,
         ),
         DataType::LargeBinary => binary::array_to_page::<i64>(
             array.as_any().downcast_ref().unwrap(),
             options,
             descriptor,
+            encoding,
         ),
         DataType::Null => {
             let array = Int32Array::new_null(DataType::Int32, array.len());
@@ -426,6 +457,7 @@ mod tests {
         nested: bool,
         version: Version,
         compression: CompressionCodec,
+        encoding: Encoding,
     ) -> Result<()> {
         let (array, statistics) = if nested {
             (
@@ -462,7 +494,7 @@ mod tests {
             std::iter::once(Result::Ok(DynIter::new(std::iter::once(Ok(DynIter::new(
                 std::iter::once(array.as_ref())
                     .zip(parquet_schema.columns().to_vec().into_iter())
-                    .map(|(array, descriptor)| array_to_page(array, descriptor, options)),
+                    .map(|(array, descriptor)| array_to_page(array, descriptor, options, encoding)),
             ))))));
 
         let mut writer = Cursor::new(vec![]);
@@ -485,116 +517,289 @@ mod tests {
 
     #[test]
     fn test_int64_optional_v1() -> Result<()> {
-        round_trip(0, true, false, Version::V1, CompressionCodec::Uncompressed)
+        round_trip(
+            0,
+            true,
+            false,
+            Version::V1,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_int64_required_v1() -> Result<()> {
-        round_trip(0, false, false, Version::V1, CompressionCodec::Uncompressed)
+        round_trip(
+            0,
+            false,
+            false,
+            Version::V1,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_int64_optional_v2() -> Result<()> {
-        round_trip(0, true, false, Version::V2, CompressionCodec::Uncompressed)
+        round_trip(
+            0,
+            true,
+            false,
+            Version::V2,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_int64_optional_v2_compressed() -> Result<()> {
-        round_trip(0, true, false, Version::V2, CompressionCodec::Snappy)
+        round_trip(
+            0,
+            true,
+            false,
+            Version::V2,
+            CompressionCodec::Snappy,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_utf8_optional_v1() -> Result<()> {
-        round_trip(2, true, false, Version::V1, CompressionCodec::Uncompressed)
+        round_trip(
+            2,
+            true,
+            false,
+            Version::V1,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_utf8_required_v1() -> Result<()> {
-        round_trip(2, false, false, Version::V1, CompressionCodec::Uncompressed)
+        round_trip(
+            2,
+            false,
+            false,
+            Version::V1,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_utf8_optional_v2() -> Result<()> {
-        round_trip(2, true, false, Version::V2, CompressionCodec::Uncompressed)
+        round_trip(
+            2,
+            true,
+            false,
+            Version::V2,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_utf8_required_v2() -> Result<()> {
-        round_trip(2, false, false, Version::V2, CompressionCodec::Uncompressed)
+        round_trip(
+            2,
+            false,
+            false,
+            Version::V2,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_utf8_optional_v2_compressed() -> Result<()> {
-        round_trip(2, true, false, Version::V2, CompressionCodec::Snappy)
+        round_trip(
+            2,
+            true,
+            false,
+            Version::V2,
+            CompressionCodec::Snappy,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_utf8_required_v2_compressed() -> Result<()> {
-        round_trip(2, false, false, Version::V2, CompressionCodec::Snappy)
+        round_trip(
+            2,
+            false,
+            false,
+            Version::V2,
+            CompressionCodec::Snappy,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_bool_optional_v1() -> Result<()> {
-        round_trip(3, true, false, Version::V1, CompressionCodec::Uncompressed)
+        round_trip(
+            3,
+            true,
+            false,
+            Version::V1,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_bool_required_v1() -> Result<()> {
-        round_trip(3, false, false, Version::V1, CompressionCodec::Uncompressed)
+        round_trip(
+            3,
+            false,
+            false,
+            Version::V1,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_bool_optional_v2_uncompressed() -> Result<()> {
-        round_trip(3, true, false, Version::V2, CompressionCodec::Uncompressed)
+        round_trip(
+            3,
+            true,
+            false,
+            Version::V2,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_bool_required_v2_uncompressed() -> Result<()> {
-        round_trip(3, false, false, Version::V2, CompressionCodec::Uncompressed)
+        round_trip(
+            3,
+            false,
+            false,
+            Version::V2,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_bool_required_v2_compressed() -> Result<()> {
-        round_trip(3, false, false, Version::V2, CompressionCodec::Snappy)
+        round_trip(
+            3,
+            false,
+            false,
+            Version::V2,
+            CompressionCodec::Snappy,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_list_int64_optional_v2() -> Result<()> {
-        round_trip(0, true, true, Version::V2, CompressionCodec::Uncompressed)
+        round_trip(
+            0,
+            true,
+            true,
+            Version::V2,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_list_int64_optional_v1() -> Result<()> {
-        round_trip(0, true, true, Version::V1, CompressionCodec::Uncompressed)
+        round_trip(
+            0,
+            true,
+            true,
+            Version::V1,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_list_bool_optional_v2() -> Result<()> {
-        round_trip(4, true, true, Version::V2, CompressionCodec::Uncompressed)
+        round_trip(
+            4,
+            true,
+            true,
+            Version::V2,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_list_bool_optional_v1() -> Result<()> {
-        round_trip(4, true, true, Version::V1, CompressionCodec::Uncompressed)
+        round_trip(
+            4,
+            true,
+            true,
+            Version::V1,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_list_utf8_optional_v2() -> Result<()> {
-        round_trip(5, true, true, Version::V2, CompressionCodec::Uncompressed)
+        round_trip(
+            5,
+            true,
+            true,
+            Version::V2,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_list_utf8_optional_v1() -> Result<()> {
-        round_trip(5, true, true, Version::V1, CompressionCodec::Uncompressed)
+        round_trip(
+            5,
+            true,
+            true,
+            Version::V1,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_list_large_binary_optional_v2() -> Result<()> {
-        round_trip(6, true, true, Version::V2, CompressionCodec::Uncompressed)
+        round_trip(
+            6,
+            true,
+            true,
+            Version::V2,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
     }
 
     #[test]
     fn test_list_large_binary_optional_v1() -> Result<()> {
-        round_trip(6, true, true, Version::V1, CompressionCodec::Uncompressed)
+        round_trip(
+            6,
+            true,
+            true,
+            Version::V1,
+            CompressionCodec::Uncompressed,
+            Encoding::Plain,
+        )
+    }
+
+    #[test]
+    fn test_utf8_optional_v2_delta() -> Result<()> {
+        round_trip(
+            2,
+            true,
+            false,
+            Version::V2,
+            CompressionCodec::Uncompressed,
+            Encoding::DeltaLengthByteArray,
+        )
     }
 }
diff --git a/src/io/parquet/write/primitive/basic.rs b/src/io/parquet/write/primitive/basic.rs
index ddb2be6010..94be5d72e6 100644
--- a/src/io/parquet/write/primitive/basic.rs
+++ b/src/io/parquet/write/primitive/basic.rs
@@ -1,6 +1,7 @@
 use parquet2::{
     metadata::ColumnDescriptor,
     read::CompressedPage,
+    schema::Encoding,
     statistics::{serialize_statistics, ParquetStatistics, PrimitiveStatistics, Statistics},
     types::NativeType,
     write::WriteOptions,
@@ -84,6 +85,7 @@ where
         statistics,
         descriptor,
         options,
+        Encoding::Plain,
     )
 }
 
diff --git a/src/io/parquet/write/primitive/nested.rs b/src/io/parquet/write/primitive/nested.rs
index 4babb4c3af..fafed4ac2c 100644
--- a/src/io/parquet/write/primitive/nested.rs
+++ b/src/io/parquet/write/primitive/nested.rs
@@ -1,3 +1,4 @@
+use parquet2::schema::Encoding;
 use parquet2::{
     metadata::ColumnDescriptor, read::CompressedPage, types::NativeType, write::WriteOptions,
 };
@@ -61,5 +62,6 @@ where
         statistics,
         descriptor,
         options,
+        Encoding::Plain,
     )
 }
diff --git a/src/io/parquet/write/record_batch.rs b/src/io/parquet/write/record_batch.rs
index 09597d945f..117cdc009d 100644
--- a/src/io/parquet/write/record_batch.rs
+++ b/src/io/parquet/write/record_batch.rs
@@ -1,5 +1,6 @@
 use super::{
-    array_to_page, to_parquet_schema, DynIter, RowGroupIter, SchemaDescriptor, WriteOptions,
+    array_to_page, to_parquet_schema, DynIter, Encoding, RowGroupIter, SchemaDescriptor,
+    WriteOptions,
 };
 use crate::{
     datatypes::Schema,
@@ -14,17 +15,26 @@ pub struct RowGroupIterator<I: Iterator<Item = Result<RecordBatch>>> {
     iter: I,
     options: WriteOptions,
     parquet_schema: SchemaDescriptor,
+    encodings: Vec<Encoding>,
 }
 
 impl<I: Iterator<Item = Result<RecordBatch>>> RowGroupIterator<I> {
     /// Creates a new [`RowGroupIterator`] from an iterator over [`RecordBatch`].
-    pub fn try_new(iter: I, schema: &Schema, options: WriteOptions) -> Result<Self> {
+    pub fn try_new(
+        iter: I,
+        schema: &Schema,
+        options: WriteOptions,
+        encodings: Vec<Encoding>,
+    ) -> Result<Self> {
+        assert_eq!(schema.fields().len(), encodings.len());
+
         let parquet_schema = to_parquet_schema(schema)?;
 
         Ok(Self {
             iter,
             options,
             parquet_schema,
+            encodings,
         })
     }
 
@@ -40,16 +50,20 @@ impl<I: Iterator<Item = Result<RecordBatch>>> Iterator for RowGroupIterator<I> {
         let options = self.options;
 
         self.iter.next().map(|batch| {
-            let columns = batch?.columns().to_vec();
+            let batch = batch?;
+            let columns = batch.columns().to_vec();
+            let encodings = self.encodings.clone();
             Ok(DynIter::new(
                 columns
                     .into_iter()
                     .zip(self.parquet_schema.columns().to_vec().into_iter())
-                    .map(move |(array, type_)| {
+                    .zip(encodings.into_iter())
+                    .map(move |((array, type_), encoding)| {
                         Ok(DynIter::new(std::iter::once(array_to_page(
                             array.as_ref(),
                             type_,
                             options,
+                            encoding,
                         ))))
                     }),
             ))
diff --git a/src/io/parquet/write/utf8/basic.rs b/src/io/parquet/write/utf8/basic.rs
index 1b86183f9d..af501f1847 100644
--- a/src/io/parquet/write/utf8/basic.rs
+++ b/src/io/parquet/write/utf8/basic.rs
@@ -1,15 +1,16 @@
 use parquet2::{
+    encoding::Encoding,
     metadata::ColumnDescriptor,
     read::CompressedPage,
     statistics::{serialize_statistics, BinaryStatistics, ParquetStatistics, Statistics},
     write::WriteOptions,
 };
 
-use super::super::binary::ord_binary;
+use super::super::binary::{encode_delta, ord_binary};
 use super::super::utils;
 use crate::{
     array::{Array, Offset, Utf8Array},
-    error::Result,
+    error::{ArrowError, Result},
     io::parquet::read::is_type_nullable,
 };
 
@@ -41,6 +42,7 @@ pub fn array_to_page<O: Offset>(
     array: &Utf8Array<O>,
     options: WriteOptions,
     descriptor: ColumnDescriptor,
+    encoding: Encoding,
 ) -> Result<CompressedPage> {
     let validity = array.validity();
     let is_optional = is_type_nullable(descriptor.type_());
@@ -56,7 +58,23 @@ pub fn array_to_page<O: Offset>(
 
     let definition_levels_byte_length = buffer.len();
 
-    encode_plain(array, is_optional, &mut buffer);
+    match encoding {
+        Encoding::Plain => encode_plain(array, is_optional, &mut buffer),
+        Encoding::DeltaLengthByteArray => encode_delta(
+            array.values(),
+            array.offsets(),
+            array.validity(),
+            is_optional,
+            &mut buffer,
+        ),
+        _ => {
+            return Err(ArrowError::InvalidArgumentError(format!(
+                "Datatype {:?} cannot be encoded by {:?} encoding",
+                array.data_type(),
+                encoding
+            )))
+        }
+    }
 
     let uncompressed_page_size = buffer.len();
 
@@ -78,6 +96,7 @@ pub fn array_to_page<O: Offset>(
         statistics,
         descriptor,
         options,
+        encoding,
     )
 }
 
diff --git a/src/io/parquet/write/utf8/nested.rs b/src/io/parquet/write/utf8/nested.rs
index 6c3b4d06b6..efc75b8aa8 100644
--- a/src/io/parquet/write/utf8/nested.rs
+++ b/src/io/parquet/write/utf8/nested.rs
@@ -1,3 +1,4 @@
+use parquet2::schema::Encoding;
 use parquet2::{metadata::ColumnDescriptor, read::CompressedPage, write::WriteOptions};
 
 use super::super::{levels, utils};
@@ -55,5 +56,6 @@ where
         statistics,
         descriptor,
         options,
+        Encoding::Plain,
     )
 }
diff --git a/src/io/parquet/write/utils.rs b/src/io/parquet/write/utils.rs
index a8c403927d..bdf247f671 100644
--- a/src/io/parquet/write/utils.rs
+++ b/src/io/parquet/write/utils.rs
@@ -69,12 +69,13 @@ pub fn build_plain_page(
     statistics: Option<ParquetStatistics>,
     descriptor: ColumnDescriptor,
     options: WriteOptions,
+    encoding: Encoding,
 ) -> Result<CompressedPage> {
     match options.version {
         Version::V1 => {
             let header = PageHeader::V1(DataPageHeader {
                 num_values: len as i32,
-                encoding: Encoding::Plain,
+                encoding,
                 definition_level_encoding: Encoding::Rle,
                 repetition_level_encoding: Encoding::Rle,
                 statistics,
@@ -92,7 +93,7 @@ pub fn build_plain_page(
         Version::V2 => {
             let header = PageHeader::V2(DataPageHeaderV2 {
                 num_values: len as i32,
-                encoding: Encoding::Plain,
+                encoding,
                 num_nulls: null_count as i32,
                 num_rows: len as i32,
                 definition_levels_byte_length: definition_levels_byte_length as i32,
@@ -142,3 +143,44 @@ pub fn compress(
         buffer
     })
 }
+
+/// Auxiliary iterator adapter to declare the size hint of an iterator.
+pub(super) struct ExactSizedIter<T, I: Iterator<Item = T>> {
+    iter: I,
+    remaining: usize,
+}
+
+impl<T, I: Iterator<Item = T> + Clone> Clone for ExactSizedIter<T, I> {
+    fn clone(&self) -> Self {
+        Self {
+            iter: self.iter.clone(),
+            remaining: self.remaining,
+        }
+    }
+}
+
+impl<T, I: Iterator<Item = T>> ExactSizedIter<T, I> {
+    pub fn new(iter: I, length: usize) -> Self {
+        Self {
+            iter,
+            remaining: length,
+        }
+    }
+}
+
+impl<T, I: Iterator<Item = T>> Iterator for ExactSizedIter<T, I> {
+    type Item = T;
+
+    #[inline]
+    fn next(&mut self) -> Option<Self::Item> {
+        self.iter.next().map(|x| {
+            self.remaining -= 1;
+            x
+        })
+    }
+
+    #[inline]
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (self.remaining, Some(self.remaining))
+    }
+}


Test patch:
diff --git a/arrow-parquet-integration-testing/main.py b/arrow-parquet-integration-testing/main.py
index f57bc6ae6d..41e7cdefde 100644
--- a/arrow-parquet-integration-testing/main.py
+++ b/arrow-parquet-integration-testing/main.py
@@ -9,11 +9,28 @@ def get_file_path(file: str):
     return f"../testing/arrow-testing/data/arrow-ipc-stream/integration/1.0.0-littleendian/{file}.arrow_file"
 
 
-def _prepare(file: str, version: str):
+def _prepare(file: str, version: str, encoding_utf8: str, projection=None):
     write = f"{file}.parquet"
-    subprocess.call(
-        ["cargo", "run", "--", "--json", file, "--output", write, "--version", version]
-    )
+
+    args = [
+        "cargo",
+        "run",
+        "--",
+        "--json",
+        file,
+        "--output",
+        write,
+        "--version",
+        version,
+        "--encoding-utf8",
+        encoding_utf8,
+    ]
+
+    if projection:
+        projection = list(map(str, projection))
+        args += ["--projection", ",".join(projection)]
+
+    subprocess.call(args)
     return write
 
 
@@ -36,26 +53,35 @@ non_native_types = [
 ]
 
 
-for version in ["1", "2"]:
-    for file in [
-        "generated_primitive",
-        "generated_primitive_no_batches",
-        "generated_primitive_zerolength",
-        "generated_null",
-        "generated_null_trivial",
-        "generated_primitive_large_offsets",
-        "generated_datetime",
-        "generated_decimal",
-        "generated_interval",
-        # requires writing Dictionary
-        # "generated_dictionary",
-        # requires writing Struct
-        # "generated_duplicate_fieldnames",
-        # requires writing un-nested List
-        # "generated_custom_metadata",
-    ]:
+def variations():
+    for version in ["1", "2"]:
+        for file in [
+            "generated_primitive",
+            "generated_primitive_no_batches",
+            "generated_primitive_zerolength",
+            "generated_null",
+            "generated_null_trivial",
+            "generated_primitive_large_offsets",
+            "generated_datetime",
+            "generated_decimal",
+            "generated_interval",
+            # requires writing Dictionary
+            # "generated_dictionary",
+            # requires writing Struct
+            # "generated_duplicate_fieldnames",
+            # requires writing un-nested List
+            # "generated_custom_metadata",
+        ]:
+            # pyarrow does not support decoding "delta"-encoded values.
+            # for encoding in ["plain", "delta"]:
+            for encoding in ["plain"]:
+                yield (version, file, encoding)
+
+
+if __name__ == "__main__":
+    for (version, file, encoding_utf8) in variations():
         expected = _expected(file)
-        path = _prepare(file, version)
+        path = _prepare(file, version, encoding_utf8)
 
         table = pq.read_table(path)
         os.remove(path)
diff --git a/arrow-parquet-integration-testing/main_spark.py b/arrow-parquet-integration-testing/main_spark.py
new file mode 100644
index 0000000000..3268d0613d
--- /dev/null
+++ b/arrow-parquet-integration-testing/main_spark.py
@@ -0,0 +1,32 @@
+"""
+Verifies that spark can correctly read a delta-encoded utf8 column written by arrow2.
+"""
+import os
+import pyspark.sql
+
+from main import _prepare, _expected
+
+_file = "generated_primitive"
+_version = "2"
+_encoding = "delta"
+column = ("utf8_nullable", 24)
+
+expected = _expected(_file)
+expected = next(c for i, c in enumerate(expected) if i == column[1])
+expected = expected.combine_chunks().tolist()
+
+path = _prepare(_file, _version, _encoding, [column[1]])
+
+spark = pyspark.sql.SparkSession.builder.config(
+    # see https://stackoverflow.com/a/62024670/931303
+    "spark.sql.parquet.enableVectorizedReader",
+    "false",
+).getOrCreate()
+
+df = spark.read.parquet(path)
+
+r = df.select(column[0]).collect()
+os.remove(path)
+
+result = [r[column[0]] for r in r]
+assert expected == result
diff --git a/arrow-parquet-integration-testing/src/main.rs b/arrow-parquet-integration-testing/src/main.rs
index 2ac77de788..7304eb0dd4 100644
--- a/arrow-parquet-integration-testing/src/main.rs
+++ b/arrow-parquet-integration-testing/src/main.rs
@@ -1,8 +1,10 @@
 use std::fs::File;
+use std::sync::Arc;
 use std::{collections::HashMap, convert::TryFrom, io::Read};
 
+use arrow2::datatypes::DataType;
 use arrow2::error::Result;
-use arrow2::io::parquet::write::RowGroupIterator;
+use arrow2::io::parquet::write::{Encoding, RowGroupIterator};
 use arrow2::io::{
     json_integration::ArrowJson,
     parquet::write::{write_file, CompressionCodec, Version, WriteOptions},
@@ -68,6 +70,18 @@ fn main() -> Result<()> {
                 .required(true)
                 .takes_value(true),
         )
+        .arg(
+            Arg::with_name("projection")
+                .long("projection")
+                .required(false)
+                .takes_value(true),
+        )
+        .arg(
+            Arg::with_name("encoding-utf8")
+                .long("encoding-utf8")
+                .required(true)
+                .takes_value(true),
+        )
         .get_matches();
     let json_file = matches
         .value_of("json")
@@ -78,9 +92,60 @@ fn main() -> Result<()> {
     let version = matches
         .value_of("version")
         .expect("must provide version of parquet");
+    let projection = matches.value_of("projection");
+    let utf8_encoding = matches
+        .value_of("encoding-utf8")
+        .expect("must provide utf8 type encoding");
+
+    let projection = projection.map(|x| {
+        x.split(',')
+            .map(|x| x.parse::<usize>().unwrap())
+            .collect::<Vec<_>>()
+    });
 
     let (schema, batches) = read_gzip_json("1.0.0-littleendian", json_file);
 
+    let schema = if let Some(projection) = &projection {
+        let fields = schema
+            .fields()
+            .iter()
+            .enumerate()
+            .filter_map(|(i, f)| {
+                if projection.contains(&i) {
+                    Some(f.clone())
+                } else {
+                    None
+                }
+            })
+            .collect::<Vec<_>>();
+        Schema::new(fields)
+    } else {
+        schema
+    };
+
+    let batches = if let Some(projection) = &projection {
+        batches
+            .iter()
+            .map(|batch| {
+                let columns = batch
+                    .columns()
+                    .iter()
+                    .enumerate()
+                    .filter_map(|(i, f)| {
+                        if projection.contains(&i) {
+                            Some(f.clone())
+                        } else {
+                            None
+                        }
+                    })
+                    .collect();
+                RecordBatch::try_new(Arc::new(schema.clone()), columns).unwrap()
+            })
+            .collect::<Vec<_>>()
+    } else {
+        batches
+    };
+
     let version = if version == "1" {
         Version::V1
     } else {
@@ -93,7 +158,23 @@ fn main() -> Result<()> {
         version,
     };
 
-    let row_groups = RowGroupIterator::try_new(batches.into_iter().map(Ok), &schema, options)?;
+    let encodings = schema
+        .fields()
+        .iter()
+        .map(|x| match x.data_type() {
+            DataType::Utf8 | DataType::LargeUtf8 => {
+                if utf8_encoding == "delta" {
+                    Encoding::DeltaLengthByteArray
+                } else {
+                    Encoding::Plain
+                }
+            }
+            _ => Encoding::Plain,
+        })
+        .collect();
+
+    let row_groups =
+        RowGroupIterator::try_new(batches.into_iter().map(Ok), &schema, options, encodings)?;
     let parquet_schema = row_groups.parquet_schema().clone();
 
     let mut writer = File::create(write_path)?;