Add ScalarValue::RunEndEncoded variant

datafusion/common/src/cast.rs

@@ -25,8 +25,9 @@ use arrow::array::{
     BinaryViewArray, Decimal32Array, Decimal64Array, DurationMicrosecondArray,
     DurationMillisecondArray, DurationNanosecondArray, DurationSecondArray, Float16Array,
     Int8Array, Int16Array, LargeBinaryArray, LargeListViewArray, LargeStringArray,
-    ListViewArray, StringViewArray, UInt16Array,
+    ListViewArray, RunArray, StringViewArray, UInt16Array,
 };
+use arrow::datatypes::RunEndIndexType;
 use arrow::{
     array::{
         Array, BinaryArray, BooleanArray, Date32Array, Date64Array, Decimal128Array,
@@ -334,3 +335,8 @@ pub fn as_list_view_array(array: &dyn Array) -> Result<&ListViewArray> {
 pub fn as_large_list_view_array(array: &dyn Array) -> Result<&LargeListViewArray> {
     Ok(downcast_value!(array, LargeListViewArray))
 }
+
+// Downcast Array to RunArray
+pub fn as_run_array<T: RunEndIndexType>(array: &dyn Array) -> Result<&RunArray<T>> {
+    Ok(downcast_value!(array, RunArray, T))
+}

datafusion/proto-common/proto/datafusion_common.proto

@@ -183,6 +183,11 @@ message Map {
   bool keys_sorted = 2;
 }
 
+message RunEndEncoded {
+  Field run_ends_field = 1;
+  Field value_field = 2;
+}
+
 enum UnionMode{
   sparse = 0;
   dense = 1;
@@ -236,6 +241,12 @@ message ScalarDictionaryValue {
   ScalarValue value = 2;
 }
 
+message ScalarRunEndEncodedValue {
+  Field run_ends_field = 1;
+  Field values_field = 2;
+  ScalarValue value = 3;
+}
+
 message IntervalDayTimeValue {
   int32 days = 1;
   int32 milliseconds = 2;
@@ -321,6 +332,8 @@ message ScalarValue{
     IntervalMonthDayNanoValue interval_month_day_nano = 31;
     ScalarFixedSizeBinary fixed_size_binary_value = 34;
     UnionValue union_value = 42;
+
+    ScalarRunEndEncodedValue run_end_encoded_value = 45;
   }
 }
 
@@ -389,6 +402,7 @@ message ArrowType{
     Union UNION = 29;
     Dictionary DICTIONARY = 30;
     Map MAP = 33;
+    RunEndEncoded RUN_END_ENCODED = 42;
   }
 }
 

datafusion/proto-common/src/from_proto/mod.rs

@@ -326,6 +326,19 @@ impl TryFrom<&protobuf::arrow_type::ArrowTypeEnum> for DataType {
                 let keys_sorted = map.keys_sorted;
                 DataType::Map(Arc::new(field), keys_sorted)
             }
+            arrow_type::ArrowTypeEnum::RunEndEncoded(run_end_encoded) => {
+                let run_ends_field: Field = run_end_encoded
+                    .as_ref()
+                    .run_ends_field
+                    .as_deref()
+                    .required("run_ends_field")?;
+                let value_field: Field = run_end_encoded
+                    .as_ref()
+                    .value_field
+                    .as_deref()
+                    .required("value_field")?;
+                DataType::RunEndEncoded(run_ends_field.into(), value_field.into())
+            }
         })
     }
 }
@@ -578,6 +591,32 @@ impl TryFrom<&protobuf::ScalarValue> for ScalarValue {
 
                 Self::Dictionary(Box::new(index_type), Box::new(value))
             }
+            Value::RunEndEncodedValue(v) => {
+                let run_ends_field: Field = v
+                    .run_ends_field
+                    .as_ref()
+                    .ok_or_else(|| Error::required("run_ends_field"))?
+                    .try_into()?;
+
+                let values_field: Field = v
+                    .values_field
+                    .as_ref()
+                    .ok_or_else(|| Error::required("values_field"))?
+                    .try_into()?;
+
+                let value: Self = v
+                    .value
+                    .as_ref()
+                    .ok_or_else(|| Error::required("value"))?
+                    .as_ref()
+                    .try_into()?;
+
+                Self::RunEndEncoded(
+                    run_ends_field.into(),
+                    values_field.into(),
+                    Box::new(value),
+                )
+            }
             Value::BinaryValue(v) => Self::Binary(Some(v.clone())),
             Value::BinaryViewValue(v) => Self::BinaryView(Some(v.clone())),
             Value::LargeBinaryValue(v) => Self::LargeBinary(Some(v.clone())),