feat(common): introduce Serial as Scalar and ScalarRef (#8327)

src/common/src/array/mod.rs

@@ -30,6 +30,7 @@ mod jsonb_array;
 pub mod list_array;
 mod macros;
 mod primitive_array;
+pub mod serial_array;
 pub mod stream_chunk;
 mod stream_chunk_iter;
 pub mod struct_array;

src/common/src/array/serial_array.rs

@@ -0,0 +1,86 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::hash::Hash;
+
+use postgres_types::{ToSql as _, Type};
+use serde::{Serialize, Serializer};
+
+use crate::types::{Scalar, ScalarRef};
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Ord, PartialOrd)]
+pub struct Serial(i64);
+
+impl Serial {
+    #[inline]
+    pub fn into_inner(self) -> i64 {
+        self.0
+    }
+}
+
+impl Serialize for Serial {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_i64(self.0)
+    }
+}
+
+impl crate::types::to_text::ToText for Serial {
+    fn write<W: std::fmt::Write>(&self, f: &mut W) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+
+    fn write_with_type<W: std::fmt::Write>(
+        &self,
+        _ty: &crate::types::DataType,
+        f: &mut W,
+    ) -> std::fmt::Result {
+        self.write(f)
+    }
+}
+
+impl crate::types::to_binary::ToBinary for Serial {
+    fn to_binary_with_type(
+        &self,
+        _ty: &crate::types::DataType,
+    ) -> crate::error::Result<Option<bytes::Bytes>> {
+        let mut output = bytes::BytesMut::new();
+        self.0.to_sql(&Type::ANY, &mut output).unwrap();
+        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/types/mod.rs

@@ -65,6 +65,7 @@ use strum_macros::EnumDiscriminants;
 use self::struct_type::StructType;
 use self::to_binary::ToBinary;
 use self::to_text::ToText;
+use crate::array::serial_array::Serial;
 use crate::array::{
     read_interval_unit, ArrayBuilderImpl, JsonbRef, JsonbVal, ListRef, ListValue,
     PrimitiveArrayItemType, StructRef, StructValue,
@@ -457,6 +458,7 @@ macro_rules! for_all_scalar_variants {
             { Int16, int16, i16, i16 },
             { Int32, int32, i32, i32 },
             { Int64, int64, i64, i64 },
+            { Serial, serial, Serial, Serial },
             { Float32, float32, OrderedF32, OrderedF32 },
             { Float64, float64, OrderedF64, OrderedF64 },
             { Utf8, utf8, Box<str>, &'scalar str },
@@ -836,6 +838,7 @@ impl ScalarRefImpl<'_> {
             Self::Int16(v) => v.serialize(ser)?,
             Self::Int32(v) => v.serialize(ser)?,
             Self::Int64(v) => v.serialize(ser)?,
+            Self::Serial(v) => v.serialize(ser)?,
             Self::Float32(v) => v.serialize(ser)?,
             Self::Float64(v) => v.serialize(ser)?,
             Self::Utf8(v) => v.serialize(ser)?,

src/common/src/types/to_binary.rs

@@ -83,6 +83,7 @@ impl ToBinary for ScalarRefImpl<'_> {
             ScalarRefImpl::Int16(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Int32(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Int64(v) => v.to_binary_with_type(ty),
+            ScalarRefImpl::Serial(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Float32(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Float64(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Utf8(v) => v.to_binary_with_type(ty),

src/common/src/util/value_encoding/mod.rs

@@ -70,6 +70,7 @@ fn serialize_scalar(value: ScalarRefImpl<'_>, buf: &mut impl BufMut) {
         ScalarRefImpl::Int16(v) => buf.put_i16_le(v),
         ScalarRefImpl::Int32(v) => buf.put_i32_le(v),
         ScalarRefImpl::Int64(v) => buf.put_i64_le(v),
+        ScalarRefImpl::Serial(v) => buf.put_i64_le(v.into_inner()),
         ScalarRefImpl::Float32(v) => buf.put_f32_le(v.into_inner()),
         ScalarRefImpl::Float64(v) => buf.put_f64_le(v.into_inner()),
         ScalarRefImpl::Utf8(v) => serialize_str(v.as_bytes(), buf),