feat(common): introduce SerialArrayBuilder and SerialArray

proto/data.proto

@@ -97,6 +97,7 @@ enum ArrayType {
   LIST = 14;
   BYTEA = 15;
   JSONB = 16;
+  SERIAL = 17;
 }
 
 message Array {

src/common/src/array/arrow.rs

@@ -444,6 +444,7 @@ impl From<&ListArray> for arrow_array::ListArray {
                 |b, v| b.append_option(v.map(|d| d.into_arrow())),
             ),
             ArrayImpl::Jsonb(_) => todo!("list of jsonb"),
+            ArrayImpl::Serial(_) => todo!("list of serial"),
             ArrayImpl::Struct(_) => todo!("list of struct"),
             ArrayImpl::List(_) => todo!("list of list"),
             ArrayImpl::Bytea(a) => build(

src/common/src/array/mod.rs

@@ -65,6 +65,7 @@ pub use utf8_array::*;
 pub use vis::{Vis, VisRef};
 
 pub use self::error::ArrayError;
+use crate::array::serial_array::{Serial, SerialArray, SerialArrayBuilder};
 use crate::buffer::Bitmap;
 use crate::types::*;
 use crate::util::iter_util::ZipEqFast;
@@ -347,6 +348,7 @@ macro_rules! for_all_variants {
             { NaiveDateTime, naivedatetime, NaiveDateTimeArray, NaiveDateTimeArrayBuilder },
             { NaiveTime, naivetime, NaiveTimeArray, NaiveTimeArrayBuilder },
             { Jsonb, jsonb, JsonbArray, JsonbArrayBuilder },
+            { Serial, serial, SerialArray, SerialArrayBuilder },
             { Struct, struct, StructArray, StructArrayBuilder },
             { List, list, ListArray, ListArrayBuilder },
             { Bytea, bytea, BytesArray, BytesArrayBuilder}
@@ -667,6 +669,9 @@ impl ArrayImpl {
             ProstArrayType::Int16 => read_numeric_array::<i16, I16ValueReader>(array, cardinality)?,
             ProstArrayType::Int32 => read_numeric_array::<i32, I32ValueReader>(array, cardinality)?,
             ProstArrayType::Int64 => read_numeric_array::<i64, I64ValueReader>(array, cardinality)?,
+            ProstArrayType::Serial => {
+                read_numeric_array::<Serial, SerialValueReader>(array, cardinality)?
+            }
             ProstArrayType::Float32 => {
                 read_numeric_array::<OrderedF32, F32ValueReader>(array, cardinality)?
             }

src/common/src/array/primitive_array.rs

@@ -21,6 +21,7 @@ use risingwave_pb::common::Buffer;
 use risingwave_pb::data::{Array as ProstArray, ArrayType};
 
 use super::{Array, ArrayBuilder, ArrayResult};
+use crate::array::serial_array::Serial;
 use crate::array::{ArrayBuilderImpl, ArrayImpl, ArrayMeta};
 use crate::buffer::{Bitmap, BitmapBuilder};
 use crate::for_all_native_types;

src/common/src/array/serial_array.rs

@@ -17,11 +17,21 @@ use std::hash::Hash;
 use postgres_types::{ToSql as _, Type};
 use serde::{Serialize, Serializer};
 
-use crate::types::{Scalar, ScalarRef};
+use crate::array::{PrimitiveArray, PrimitiveArrayBuilder};
 
-#[derive(Debug, Copy, Clone, PartialEq, Eq, Ord, PartialOrd)]
+// Serial is an alias for i64
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Ord, PartialOrd, Default, Hash)]
 pub struct Serial(i64);
 
+pub type SerialArray = PrimitiveArray<Serial>;
+pub type SerialArrayBuilder = PrimitiveArrayBuilder<Serial>;
+
+impl From<i64> for Serial {
+    fn from(value: i64) -> Self {
+        Self(value)
+    }
+}
+
 impl Serial {
     #[inline]
     pub fn into_inner(self) -> i64 {
@@ -62,25 +72,3 @@ impl crate::types::to_binary::ToBinary for Serial {
         Ok(Some(output.freeze()))
     }
 }
-
-/// Implement `Scalar` for `Serial`.
-impl Scalar for Serial {
-    type ScalarRefType<'a> = Serial;
-
-    fn as_scalar_ref(&self) -> Self::ScalarRefType<'_> {
-        Serial(self.0)
-    }
-}
-
-/// Implement `ScalarRef` for `Serial`.
-impl<'a> ScalarRef<'a> for Serial {
-    type ScalarType = Serial;
-
-    fn to_owned_scalar(&self) -> Serial {
-        *self
-    }
-
-    fn hash_scalar<H: std::hash::Hasher>(&self, state: &mut H) {
-        self.0.hash(state)
-    }
-}

src/common/src/array/value_reader.rs

@@ -19,7 +19,8 @@ use byteorder::{BigEndian, ReadBytesExt};
 
 use super::ArrayResult;
 use crate::array::{
-    ArrayBuilder, BytesArrayBuilder, JsonbArrayBuilder, PrimitiveArrayItemType, Utf8ArrayBuilder,
+    ArrayBuilder, BytesArrayBuilder, JsonbArrayBuilder, PrimitiveArrayItemType, Serial,
+    Utf8ArrayBuilder,
 };
 use crate::types::{Decimal, OrderedF32, OrderedF64};
 
@@ -31,6 +32,7 @@ pub trait PrimitiveValueReader<T: PrimitiveArrayItemType> {
 pub struct I16ValueReader;
 pub struct I32ValueReader;
 pub struct I64ValueReader;
+pub struct SerialValueReader;
 pub struct F32ValueReader;
 pub struct F64ValueReader;
 
@@ -50,6 +52,7 @@ macro_rules! impl_numeric_value_reader {
 impl_numeric_value_reader!(i16, I16ValueReader, read_i16);
 impl_numeric_value_reader!(i32, I32ValueReader, read_i32);
 impl_numeric_value_reader!(i64, I64ValueReader, read_i64);
+impl_numeric_value_reader!(Serial, SerialValueReader, read_i64);
 impl_numeric_value_reader!(OrderedF32, F32ValueReader, read_f32);
 impl_numeric_value_reader!(OrderedF64, F64ValueReader, read_f64);
 

src/common/src/hash/key.rs

@@ -30,6 +30,7 @@ use std::io::{Cursor, Read};
 use chrono::{Datelike, Timelike};
 use fixedbitset::FixedBitSet;
 
+use crate::array::serial_array::Serial;
 use crate::array::{
     Array, ArrayBuilder, ArrayBuilderImpl, ArrayError, ArrayImpl, ArrayResult, DataChunk, JsonbRef,
     ListRef, StructRef,
@@ -477,6 +478,18 @@ impl<'a> HashKeySerDe<'a> for JsonbRef<'a> {
     }
 }
 
+impl<'a> HashKeySerDe<'a> for Serial {
+    type S = <i64 as HashKeySerDe<'a>>::S;
+
+    fn serialize(self) -> Self::S {
+        self.into_inner().serialize()
+    }
+
+    fn deserialize<R: Read>(source: &mut R) -> Self {
+        i64::deserialize(source).into()
+    }
+}
+
 impl<'a> HashKeySerDe<'a> for StructRef<'a> {
     type S = Vec<u8>;