Reduce the memory size of AggregateFunction state

[sundy-li]

Summary

This test code proves that DataValue has memory size 64 bytes, because DataValue is an enum struct which uses the maximum memory in each enum.

 println!("memory size -> {:?}", std::mem::size_of::<DataValue>());   // 64 
 println!("memory size -> {:?}", std::mem::size_of::<Option<u64>>()); // 16

We should introduce a generic state implementation to improve memory and efficiency.

[sundy-li]

Example for aggregate sum state.

struct AggregateSumState<T> {
    pub value: Option<T>,
}

impl<'a, T> GetState<'a, AggregateSumState<T>> for AggregateSumState<T> {}

impl<T> AggregateSumState<T>
where T: std::ops::Add<Output = T> + Clone + Copy + serde::Serialize + serde::de::DeserializeOwned
{
    #[inline(always)]
    fn add(&mut self, other: &Option<T>) {
        match (&self.value, other) {
            (Some(a), Some(b)) => self.value = Some(a.add(*b)),
            (None, c) => self.value = c.clone(),
            _ => {}
        }
    }

    #[inline(always)]
    fn merge(&mut self, other: &Self) {
        match (&self.value, &other.value) {
            (Some(a), Some(b)) => self.value = Some(a.add(*b)),
            (None, c) => self.value = c.clone(),
            _ => {}
        }
    }

    pub fn serialize(&self, writer: &mut Vec<u8>) -> Result<()> {
        serde_json::to_writer(writer, &self.value)?;
        Ok(())
    }

    pub fn deserialize<R: std::io::Read>(&mut self, reader: R) -> Result<()> {
        self.value = serde_json::from_reader(reader)?;
        Ok(())
    }
}