[FEAT] implement series serde

daft/daft.pyi

@@ -665,6 +665,9 @@ class PySeries:
     def image_resize(self, w: int, h: int) -> PySeries: ...
     def if_else(self, other: PySeries, predicate: PySeries) -> PySeries: ...
     def is_null(self) -> PySeries: ...
+    def _debug_bincode_serialize(self) -> bytes: ...
+    @staticmethod
+    def _debug_bincode_deserialize(b: bytes) -> PySeries: ...
 
 class PyTable:
     def schema(self) -> PySchema: ...

daft/series.py

@@ -527,6 +527,13 @@ def _from_arrow_table_to_series(cls, table: pa.Table, extension_type: pa.Extensi
             array = extension_type.wrap_array(array)
         return cls.from_arrow(array, name)
 
+    def _debug_bincode_serialize(self) -> bytes:
+        return self._series._debug_bincode_serialize()
+
+    @classmethod
+    def _debug_bincode_deserialize(cls, b: bytes) -> Series:
+        return Series._from_pyseries(PySeries._debug_bincode_deserialize(b))
+
 
 SomeSeriesNamespace = TypeVar("SomeSeriesNamespace", bound="SeriesNamespace")
 

src/daft-core/src/array/boolean.rs

@@ -0,0 +1,9 @@
+use crate::datatypes::BooleanArray;
+
+use super::ops::as_arrow::AsArrow;
+
+impl BooleanArray {
+    pub fn as_bitmap(&self) -> &arrow2::bitmap::Bitmap {
+        self.as_arrow().values()
+    }
+}

src/daft-core/src/array/from.rs

@@ -6,6 +6,7 @@ use crate::datatypes::{
 };
 
 use crate::array::DataArray;
+
 use common_error::{DaftError, DaftResult};
 
 impl<T: DaftNumericType> From<(&str, Box<arrow2::array::PrimitiveArray<T::Native>>)>
@@ -63,6 +64,16 @@ impl From<(&str, &[bool])> for BooleanArray {
     }
 }
 
+impl From<(&str, &[Option<bool>])> for BooleanArray {
+    fn from(item: (&str, &[Option<bool>])) -> Self {
+        let (name, slice) = item;
+        let arrow_array = Box::new(arrow2::array::BooleanArray::from_trusted_len_iter(
+            slice.iter().cloned(),
+        ));
+        DataArray::new(Field::new(name, DataType::Boolean).into(), arrow_array).unwrap()
+    }
+}
+
 impl From<(&str, arrow2::array::BooleanArray)> for BooleanArray {
     fn from(item: (&str, arrow2::array::BooleanArray)) -> Self {
         let (name, arrow_array) = item;
@@ -74,6 +85,21 @@ impl From<(&str, arrow2::array::BooleanArray)> for BooleanArray {
     }
 }
 
+impl From<(&str, arrow2::bitmap::Bitmap)> for BooleanArray {
+    fn from(item: (&str, arrow2::bitmap::Bitmap)) -> Self {
+        let (name, bitmap) = item;
+        DataArray::new(
+            Field::new(name, DataType::Boolean).into(),
+            Box::new(arrow2::array::BooleanArray::new(
+                arrow2::datatypes::DataType::Boolean,
+                bitmap,
+                None,
+            )),
+        )
+        .unwrap()
+    }
+}
+
 impl From<(&str, Box<arrow2::array::BooleanArray>)> for BooleanArray {
     fn from(item: (&str, Box<arrow2::array::BooleanArray>)) -> Self {
         let (name, arrow_array) = item;

src/daft-core/src/array/from_iter.rs

@@ -0,0 +1,43 @@
+use crate::datatypes::{BinaryArray, DaftNumericType, Field, Utf8Array};
+
+use super::DataArray;
+
+impl<T> DataArray<T>
+where
+    T: DaftNumericType,
+{
+    pub fn from_iter(
+        name: &str,
+        iter: impl Iterator<Item = Option<T::Native>> + arrow2::trusted_len::TrustedLen,
+    ) -> Self {
+        let arrow_array =
+            Box::new(arrow2::array::PrimitiveArray::<T::Native>::from_trusted_len_iter(iter));
+        DataArray::new(Field::new(name, T::get_dtype()).into(), arrow_array).unwrap()
+    }
+}
+
+impl Utf8Array {
+    pub fn from_iter<S: AsRef<str>>(
+        name: &str,
+        iter: impl Iterator<Item = Option<S>> + arrow2::trusted_len::TrustedLen,
+    ) -> Self {
+        let arrow_array = Box::new(arrow2::array::Utf8Array::<i64>::from_trusted_len_iter(iter));
+        DataArray::new(Field::new(name, crate::DataType::Utf8).into(), arrow_array).unwrap()
+    }
+}
+
+impl BinaryArray {
+    pub fn from_iter<S: AsRef<[u8]>>(
+        name: &str,
+        iter: impl Iterator<Item = Option<S>> + arrow2::trusted_len::TrustedLen,
+    ) -> Self {
+        let arrow_array = Box::new(arrow2::array::BinaryArray::<i64>::from_trusted_len_iter(
+            iter,
+        ));
+        DataArray::new(
+            Field::new(name, crate::DataType::Binary).into(),
+            arrow_array,
+        )
+        .unwrap()
+    }
+}

src/daft-core/src/array/mod.rs

@@ -1,15 +1,18 @@
+mod fixed_size_list_array;
 pub mod from;
 pub mod growable;
 pub mod iterator;
+mod list_array;
 pub mod ops;
 pub mod pseudo_arrow;
-
-mod fixed_size_list_array;
-mod list_array;
+mod serdes;
 mod struct_array;
 pub use fixed_size_list_array::FixedSizeListArray;
 pub use list_array::ListArray;
+
 pub use struct_array::StructArray;
+mod boolean;
+mod from_iter;
 
 use std::{marker::PhantomData, sync::Arc};
 

src/daft-core/src/array/serdes.rs

@@ -0,0 +1,224 @@
+use std::cell::RefCell;
+
+use serde::ser::SerializeMap;
+
+use crate::{
+    datatypes::{
+        logical::LogicalArray, BinaryArray, BooleanArray, DaftLogicalType, DaftNumericType,
+        ExtensionArray, Int64Array, NullArray, Utf8Array,
+    },
+    DataType, IntoSeries, Series,
+};
+
+#[cfg(feature = "python")]
+use crate::datatypes::PythonArray;
+
+use super::{ops::as_arrow::AsArrow, DataArray, FixedSizeListArray, ListArray, StructArray};
+
+// adapted from Polars Serdes iterator
+pub struct IterSer<I>
+where
+    I: IntoIterator,
+    <I as IntoIterator>::Item: serde::Serialize,
+{
+    iter: RefCell<Option<I>>,
+}
+
+impl<I> IterSer<I>
+where
+    I: IntoIterator,
+    <I as IntoIterator>::Item: serde::Serialize,
+{
+    fn new(iter: I) -> Self {
+        IterSer {
+            iter: RefCell::new(Some(iter)),
+        }
+    }
+}
+
+impl<I> serde::Serialize for IterSer<I>
+where
+    I: IntoIterator,
+    <I as IntoIterator>::Item: serde::Serialize,
+{
+    fn serialize<S>(
+        &self,
+        serializer: S,
+    ) -> std::result::Result<<S as serde::Serializer>::Ok, <S as serde::Serializer>::Error>
+    where
+        S: serde::Serializer,
+    {
+        let iter: I = self.iter.borrow_mut().take().unwrap();
+        serializer.collect_seq(iter)
+    }
+}
+
+impl<T: DaftNumericType> serde::Serialize for DataArray<T> {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut s = serializer.serialize_map(Some(2))?;
+        s.serialize_entry("field", self.field())?;
+        s.serialize_entry("values", &IterSer::new(self.as_arrow().iter()))?;
+        s.end()
+    }
+}
+
+impl serde::Serialize for Utf8Array {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut s = serializer.serialize_map(Some(2))?;
+        s.serialize_entry("field", self.field())?;
+        s.serialize_entry("values", &IterSer::new(self.as_arrow().iter()))?;
+        s.end()
+    }
+}
+
+impl serde::Serialize for BooleanArray {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut s = serializer.serialize_map(Some(2))?;
+        s.serialize_entry("field", self.field())?;
+        s.serialize_entry("values", &IterSer::new(self.as_arrow().iter()))?;
+        s.end()
+    }
+}
+
+impl serde::Serialize for BinaryArray {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut s = serializer.serialize_map(Some(2))?;
+        s.serialize_entry("field", self.field())?;
+        s.serialize_entry("values", &IterSer::new(self.as_arrow().iter()))?;
+        s.end()
+    }
+}
+
+impl serde::Serialize for NullArray {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut s = serializer.serialize_map(Some(2))?;
+        s.serialize_entry("field", self.field())?;
+        s.serialize_entry("values", &self.len())?;
+        s.end()
+    }
+}
+
+impl serde::Serialize for ExtensionArray {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut s = serializer.serialize_map(Some(2))?;
+        s.serialize_entry("field", self.field())?;
+        let values = if let DataType::Extension(_, inner, _) = self.data_type() {
+            Series::try_from(("physical", self.data.to_type(inner.to_arrow().unwrap()))).unwrap()
+        } else {
+            panic!("Expected Extension Type!")
+        };
+        s.serialize_entry("values", &values)?;
+        s.end()
+    }
+}
+
+#[cfg(feature = "python")]
+impl serde::Serialize for PythonArray {
+    fn serialize<S>(&self, _serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        panic!("Rust Serde is not implemented for Python Arrays")
+    }
+}
+
+impl serde::Serialize for StructArray {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut s = serializer.serialize_map(Some(2))?;
+
+        let mut values = Vec::with_capacity(self.children.len() + 1);
+        values.extend(self.children.iter().map(Some));
+
+        let validity = self
+            .validity()
+            .map(|b| BooleanArray::from(("validity", b.clone())).into_series());
+        values.push(validity.as_ref());
+
+        s.serialize_entry("field", self.field.as_ref())?;
+        s.serialize_entry("values", &values)?;
+        s.end()
+    }
+}
+
+impl serde::Serialize for ListArray {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut s = serializer.serialize_map(Some(2))?;
+        let mut values = Vec::with_capacity(3);
+
+        values.push(Some(&self.flat_child));
+
+        let arrow2_offsets = arrow2::array::Int64Array::new(
+            arrow2::datatypes::DataType::Int64,
+            self.offsets().buffer().clone(),
+            None,
+        );
+        let offsets = Int64Array::from(("offsets", Box::new(arrow2_offsets))).into_series();
+        values.push(Some(&offsets));
+
+        let validity = self
+            .validity()
+            .map(|b| BooleanArray::from(("validity", b.clone())).into_series());
+        values.push(validity.as_ref());
+
+        s.serialize_entry("field", self.field.as_ref())?;
+        s.serialize_entry("values", &values)?;
+        s.end()
+    }
+}
+
+impl serde::Serialize for FixedSizeListArray {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut s = serializer.serialize_map(Some(2))?;
+
+        let validity = self
+            .validity()
+            .map(|b| BooleanArray::from(("validity", b.clone())).into_series());
+        let values = vec![Some(&self.flat_child), validity.as_ref()];
+        s.serialize_entry("field", self.field.as_ref())?;
+        s.serialize_entry("values", &values)?;
+        s.end()
+    }
+}
+
+impl<L: DaftLogicalType> serde::Serialize for LogicalArray<L>
+where
+    <<L as DaftLogicalType>::PhysicalType as crate::datatypes::DaftDataType>::ArrayType:
+        serde::Serialize + IntoSeries,
+{
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut s = serializer.serialize_map(Some(2))?;
+        s.serialize_entry("field", self.field.as_ref())?;
+        s.serialize_entry("values", &self.physical.clone().into_series())?;
+        s.end()
+    }
+}

src/daft-core/src/datatypes/mod.rs

@@ -6,8 +6,6 @@ mod image_mode;
 mod matching;
 mod time_unit;
 
-use std::ops::{Add, Div, Mul, Rem, Sub};
-
 pub use crate::array::{DataArray, FixedSizeListArray};
 use crate::array::{ListArray, StructArray};
 use arrow2::{
@@ -21,6 +19,8 @@ pub use field::FieldID;
 pub use image_format::ImageFormat;
 pub use image_mode::ImageMode;
 use num_traits::{Bounded, Float, FromPrimitive, Num, NumCast, ToPrimitive, Zero};
+use serde::Serialize;
+use std::ops::{Add, Div, Mul, Rem, Sub};
 pub use time_unit::TimeUnit;
 
 pub mod logical;
@@ -201,6 +201,7 @@ pub trait NumericNative:
     + Bounded
     + FromPrimitive
     + ToPrimitive
+    + Serialize
 {
     type DAFTTYPE: DaftNumericType;
 }