Add Run End Encoding DataType

arrow-array/src/array/mod.rs

@@ -737,6 +737,7 @@ pub fn new_null_array(data_type: &DataType, length: usize) -> ArrayRef {
             new_null_sized_decimal(data_type, length, std::mem::size_of::<i128>())
         }
         DataType::Decimal256(_, _) => new_null_sized_decimal(data_type, length, 32),
+        DataType::RunEndEncoded(_, _) => todo!(),
     }
 }
 

arrow-data/src/data.rs

@@ -198,9 +198,9 @@ pub(crate) fn new_buffers(data_type: &DataType, capacity: usize) -> [MutableBuff
             ],
             _ => unreachable!(),
         },
-        DataType::FixedSizeList(_, _) | DataType::Struct(_) => {
-            [empty_buffer, MutableBuffer::new(0)]
-        }
+        DataType::FixedSizeList(_, _)
+        | DataType::Struct(_)
+        | DataType::RunEndEncoded(_, _) => [empty_buffer, MutableBuffer::new(0)],
         DataType::Decimal128(_, _) | DataType::Decimal256(_, _) => [
             MutableBuffer::new(capacity * mem::size_of::<u8>()),
             empty_buffer,
@@ -654,6 +654,12 @@ impl ArrayData {
             DataType::Dictionary(_, data_type) => {
                 vec![Self::new_empty(data_type)]
             }
+            DataType::RunEndEncoded(run_ends, values) => {
+                vec![
+                    Self::new_empty(run_ends.data_type()),
+                    Self::new_empty(values.data_type()),
+                ]
+            }
         };
 
         // Data was constructed correctly above
@@ -1508,6 +1514,7 @@ pub fn layout(data_type: &DataType) -> DataTypeLayout {
             // same as ListType
             DataTypeLayout::new_fixed_width(size_of::<i32>())
         }
+        DataType::RunEndEncoded(_, _) => DataTypeLayout::new_empty(), // all in child data,
     }
 }
 

arrow-data/src/equal/mod.rs

@@ -137,6 +137,7 @@ fn equal_values(
         },
         DataType::Float16 => primitive_equal::<f16>(lhs, rhs, lhs_start, rhs_start, len),
         DataType::Map(_, _) => list_equal::<i32>(lhs, rhs, lhs_start, rhs_start, len),
+        DataType::RunEndEncoded(_, _) => todo!(),
     }
 }
 

arrow-data/src/transform/mod.rs

@@ -230,6 +230,7 @@ fn build_extend(array: &ArrayData) -> Extend {
             UnionMode::Sparse => union::build_extend_sparse(array),
             UnionMode::Dense => union::build_extend_dense(array),
         },
+        DataType::RunEndEncoded(_, _) => todo!(),
     }
 }
 
@@ -281,6 +282,7 @@ fn build_extend_nulls(data_type: &DataType) -> ExtendNulls {
             UnionMode::Sparse => union::extend_nulls_sparse,
             UnionMode::Dense => union::extend_nulls_dense,
         },
+        DataType::RunEndEncoded(_, _) => todo!(),
     })
 }
 
@@ -473,6 +475,20 @@ impl<'a> MutableArrayData<'a> {
                     })
                     .collect::<Vec<_>>(),
             },
+            DataType::RunEndEncoded(_, _) => {
+                let run_ends_child = arrays
+                    .iter()
+                    .map(|array| &array.child_data()[0])
+                    .collect::<Vec<_>>();
+                let value_child = arrays
+                    .iter()
+                    .map(|array| &array.child_data()[1])
+                    .collect::<Vec<_>>();
+                vec![
+                    MutableArrayData::new(run_ends_child, false, array_capacity),
+                    MutableArrayData::new(value_child, use_nulls, array_capacity),
+                ]
+            }
             DataType::FixedSizeList(_, _) => {
                 let childs = arrays
                     .iter()

arrow-integration-test/src/datatype.rs

@@ -357,6 +357,7 @@ pub fn data_type_to_json(data_type: &DataType) -> serde_json::Value {
         DataType::Map(_, keys_sorted) => {
             json!({"name": "map", "keysSorted": keys_sorted})
         }
+        DataType::RunEndEncoded(_, _) => todo!(),
     }
 }
 

arrow-ipc/src/convert.rs

@@ -772,6 +772,7 @@ pub(crate) fn get_fb_field_type<'a>(
                 children: Some(fbb.create_vector(&children[..])),
             }
         }
+        RunEndEncoded(_, _) => todo!(),
     }
 }
 

arrow-schema/src/datatype.rs

@@ -242,6 +242,19 @@ pub enum DataType {
     /// child fields may be respectively "entries", "key", and "value", but this is
     /// not enforced.
     Map(Box<Field>, bool),
+
+    /// A run-end encoding (REE) array is a variation of run-length encoding (RLE). These
+    /// encodings are well-suited for representing data containing sequences of the
+    /// same value, called runs. Each run is represented as a value and an integer giving
+    /// the index in the array where the run ends.
+    ///
+    /// A run-end encoded array has no buffers by itself, but has two child arrays. The
+    /// first child array, called the run ends array, holds either 16, 32, or 64-bit
+    /// signed integers. The actual values of each run are held in the second child array.
+    ///
+    /// These child arrays are prescribed the standard names of "run_ends" and "values"
+    /// respectively.
+    RunEndEncoded(Box<Field>, Box<Field>),
 }
 
 /// An absolute length of time in seconds, milliseconds, microseconds or nanoseconds.
@@ -438,6 +451,10 @@ impl DataType {
                         + (std::mem::size_of::<Field>() * fields.capacity())
                 }
                 DataType::Dictionary(dt1, dt2) => dt1.size() + dt2.size(),
+                DataType::RunEndEncoded(run_ends, values) => {
+                    run_ends.size() - std::mem::size_of_val(run_ends) + values.size()
+                        - std::mem::size_of_val(values)
+                }
             }
     }
 }

arrow-schema/src/field.rs

@@ -410,6 +410,7 @@ impl Field {
             | DataType::List(_)
             | DataType::Map(_, _)
             | DataType::Dictionary(_, _)
+            | DataType::RunEndEncoded(_, _)
             | DataType::FixedSizeList(_, _)
             | DataType::FixedSizeBinary(_)
             | DataType::Utf8

parquet/src/arrow/arrow_writer/mod.rs

@@ -360,7 +360,7 @@ fn write_leaves<W: Write>(
         ArrowDataType::Float16 => Err(ParquetError::ArrowError(
             "Float16 arrays not supported".to_string(),
         )),
-        ArrowDataType::FixedSizeList(_, _) | ArrowDataType::Union(_, _, _) => {
+        ArrowDataType::FixedSizeList(_, _) | ArrowDataType::Union(_, _, _) | ArrowDataType::RunEndEncoded(_, _) => {
             Err(ParquetError::NYI(
                 format!(
                     "Attempting to write an Arrow type {:?} to parquet that is not yet implemented",

parquet/src/arrow/schema/mod.rs

@@ -507,6 +507,9 @@ fn arrow_to_parquet_type(field: &Field) -> Result<Type> {
             let dict_field = Field::new(name, *value.clone(), field.is_nullable());
             arrow_to_parquet_type(&dict_field)
         }
+        DataType::RunEndEncoded(_, _) => Err(arrow_err!(
+            "Converting RunEndEncodedType to parquet not supported",
+        )),
     }
 }
 
@@ -640,7 +643,7 @@ mod tests {
             ProjectionMask::all(),
             None,
         )
-            .unwrap();
+        .unwrap();
         assert_eq!(&arrow_fields, converted_arrow_schema.fields());
     }
 
@@ -1342,20 +1345,9 @@ mod tests {
                 ))),
                 false,
             ),
-            Field::new(
-                "decimal_int32",
-                DataType::Decimal128(8, 2),
-                false,
-            ),
-            Field::new(
-                "decimal_int64",
-                DataType::Decimal128(16, 2),
-                false,
-            ),
-            Field::new(
-                "decimal_fix_length",
-                DataType::Decimal128(30, 2),
-                false, ),
+            Field::new("decimal_int32", DataType::Decimal128(8, 2), false),
+            Field::new("decimal_int64", DataType::Decimal128(16, 2), false),
+            Field::new("decimal_fix_length", DataType::Decimal128(30, 2), false),
         ];
 
         assert_eq!(arrow_fields, converted_arrow_fields);
@@ -1491,18 +1483,9 @@ mod tests {
                 DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)),
                 false,
             ),
-            Field::new(
-                "decimal_int32",
-                DataType::Decimal128(8, 2),
-                false),
-            Field::new("decimal_int64",
-                       DataType::Decimal128(16, 2),
-                       false),
-            Field::new(
-                "decimal_fix_length",
-                DataType::Decimal128(30, 2),
-                false,
-            ),
+            Field::new("decimal_int32", DataType::Decimal128(8, 2), false),
+            Field::new("decimal_int64", DataType::Decimal128(16, 2), false),
+            Field::new("decimal_fix_length", DataType::Decimal128(30, 2), false),
         ];
         let arrow_schema = Schema::new(arrow_fields);
         let converted_arrow_schema = arrow_to_parquet_schema(&arrow_schema).unwrap();