From 9d465e5842b5f19d02138caff8d8796d605f89ac Mon Sep 17 00:00:00 2001
From: xxchan <xxchan22f@gmail.com>
Date: Sat, 20 Jul 2024 10:54:06 +0800
Subject: [PATCH] refactor(types): simplify ArrayImpl::from_protobuf

---
 src/common/src/array/mod.rs             |   1 -
 src/common/src/array/primitive_array.rs |  16 ++-
 src/common/src/array/proto_reader.rs    | 154 ++++++++----------------
 src/common/src/array/value_reader.rs    |  88 --------------
 src/common/src/types/datetime.rs        |  28 ++++-
 src/common/src/types/interval.rs        |  15 ++-
 src/common/src/types/mod.rs             |  12 +-
 src/common/src/types/scalar_impl.rs     |   2 +-
 src/common/src/types/timestamptz.rs     |  11 +-
 9 files changed, 118 insertions(+), 209 deletions(-)
 delete mode 100644 src/common/src/array/value_reader.rs

diff --git a/src/common/src/array/mod.rs b/src/common/src/array/mod.rs
index f2ae95aa71efb..89b3b06266786 100644
--- a/src/common/src/array/mod.rs
+++ b/src/common/src/array/mod.rs
@@ -35,7 +35,6 @@ mod stream_chunk_iter;
 pub mod stream_record;
 pub mod struct_array;
 mod utf8_array;
-mod value_reader;
 
 use std::convert::From;
 use std::hash::{Hash, Hasher};
diff --git a/src/common/src/array/primitive_array.rs b/src/common/src/array/primitive_array.rs
index 9ae7912e9887e..45e1bc0ddd847 100644
--- a/src/common/src/array/primitive_array.rs
+++ b/src/common/src/array/primitive_array.rs
@@ -13,9 +13,11 @@
 // limitations under the License.
 
 use std::fmt::Debug;
-use std::io::Write;
+use std::io::{Cursor, Write};
 use std::mem::size_of;
 
+use anyhow::Context;
+use byteorder::{BigEndian, ReadBytesExt};
 use risingwave_common_estimate_size::{EstimateSize, ZeroHeapSize};
 use risingwave_pb::common::buffer::CompressionType;
 use risingwave_pb::common::Buffer;
@@ -50,6 +52,7 @@ where
 
     // item methods
     fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize>;
+    fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<Self>;
 }
 
 macro_rules! impl_array_methods {
@@ -81,7 +84,7 @@ macro_rules! impl_array_methods {
 }
 
 macro_rules! impl_primitive_for_native_types {
-    ($({ $naive_type:ty, $scalar_type:ident } ),*) => {
+    ($({ $naive_type:ty, $scalar_type:ident, $read_fn:ident } ),*) => {
         $(
             impl PrimitiveArrayItemType for $naive_type {
                 impl_array_methods!($naive_type, $scalar_type, $scalar_type);
@@ -89,6 +92,12 @@ macro_rules! impl_primitive_for_native_types {
                 fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize> {
                     NativeType::to_protobuf(self, output)
                 }
+                fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<Self> {
+                    let v = cur
+                        .$read_fn::<BigEndian>()
+                        .context("failed to read value from buffer")?;
+                    Ok(v.into())
+                }
             }
         )*
     }
@@ -106,6 +115,9 @@ macro_rules! impl_primitive_for_others {
                 fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize> {
                     <$scalar_type>::to_protobuf(self, output)
                 }
+                fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<Self> {
+                    <$scalar_type>::from_protobuf(cur)
+                }
             }
         )*
     }
diff --git a/src/common/src/array/proto_reader.rs b/src/common/src/array/proto_reader.rs
index 7c3b05437770c..aa296900190df 100644
--- a/src/common/src/array/proto_reader.rs
+++ b/src/common/src/array/proto_reader.rs
@@ -16,43 +16,32 @@ use std::io::{Cursor, Read};
 
 use anyhow::Context;
 use byteorder::{BigEndian, ReadBytesExt};
-use paste::paste;
 use risingwave_pb::data::PbArrayType;
 
 use super::*;
-use crate::array::value_reader::{PrimitiveValueReader, VarSizedValueReader};
 
 impl ArrayImpl {
     pub fn from_protobuf(array: &PbArray, cardinality: usize) -> ArrayResult<Self> {
-        use crate::array::value_reader::*;
         let array = match array.array_type() {
             PbArrayType::Unspecified => unreachable!(),
-            PbArrayType::Int16 => read_numeric_array::<i16, I16ValueReader>(array, cardinality)?,
-            PbArrayType::Int32 => read_numeric_array::<i32, I32ValueReader>(array, cardinality)?,
-            PbArrayType::Int64 => read_numeric_array::<i64, I64ValueReader>(array, cardinality)?,
-            PbArrayType::Serial => {
-                read_numeric_array::<Serial, SerialValueReader>(array, cardinality)?
-            }
-            PbArrayType::Float32 => read_numeric_array::<F32, F32ValueReader>(array, cardinality)?,
-            PbArrayType::Float64 => read_numeric_array::<F64, F64ValueReader>(array, cardinality)?,
+            PbArrayType::Int16 => read_primitive_array::<i16>(array, cardinality)?,
+            PbArrayType::Int32 => read_primitive_array::<i32>(array, cardinality)?,
+            PbArrayType::Int64 => read_primitive_array::<i64>(array, cardinality)?,
+            PbArrayType::Serial => read_primitive_array::<Serial>(array, cardinality)?,
+            PbArrayType::Float32 => read_primitive_array::<F32>(array, cardinality)?,
+            PbArrayType::Float64 => read_primitive_array::<F64>(array, cardinality)?,
             PbArrayType::Bool => read_bool_array(array, cardinality)?,
-            PbArrayType::Utf8 => {
-                read_string_array::<Utf8ArrayBuilder, Utf8ValueReader>(array, cardinality)?
-            }
-            PbArrayType::Decimal => {
-                read_numeric_array::<Decimal, DecimalValueReader>(array, cardinality)?
-            }
-            PbArrayType::Date => read_date_array(array, cardinality)?,
-            PbArrayType::Time => read_time_array(array, cardinality)?,
-            PbArrayType::Timestamp => read_timestamp_array(array, cardinality)?,
-            PbArrayType::Timestamptz => read_timestamptz_array(array, cardinality)?,
-            PbArrayType::Interval => read_interval_array(array, cardinality)?,
+            PbArrayType::Utf8 => read_string_array::<Utf8ValueReader>(array, cardinality)?,
+            PbArrayType::Decimal => read_primitive_array::<Decimal>(array, cardinality)?,
+            PbArrayType::Date => read_primitive_array::<Date>(array, cardinality)?,
+            PbArrayType::Time => read_primitive_array::<Time>(array, cardinality)?,
+            PbArrayType::Timestamp => read_primitive_array::<Timestamp>(array, cardinality)?,
+            PbArrayType::Timestamptz => read_primitive_array::<Timestamptz>(array, cardinality)?,
+            PbArrayType::Interval => read_primitive_array::<Interval>(array, cardinality)?,
             PbArrayType::Jsonb => JsonbArray::from_protobuf(array)?,
             PbArrayType::Struct => StructArray::from_protobuf(array)?,
             PbArrayType::List => ListArray::from_protobuf(array)?,
-            PbArrayType::Bytea => {
-                read_string_array::<BytesArrayBuilder, BytesValueReader>(array, cardinality)?
-            }
+            PbArrayType::Bytea => read_string_array::<BytesValueReader>(array, cardinality)?,
             PbArrayType::Int256 => Int256Array::from_protobuf(array, cardinality)?,
         };
         Ok(array)
@@ -62,7 +51,7 @@ impl ArrayImpl {
 // TODO: Use techniques like apache arrow flight RPC to eliminate deserialization.
 // https://arrow.apache.org/docs/format/Flight.html
 
-fn read_numeric_array<T: PrimitiveArrayItemType, R: PrimitiveValueReader<T>>(
+fn read_primitive_array<T: PrimitiveArrayItemType>(
     array: &PbArray,
     cardinality: usize,
 ) -> ArrayResult<ArrayImpl> {
@@ -78,7 +67,7 @@ fn read_numeric_array<T: PrimitiveArrayItemType, R: PrimitiveValueReader<T>>(
     let mut cursor = Cursor::new(buf);
     for not_null in bitmap.iter() {
         if not_null {
-            let v = R::read(&mut cursor)?;
+            let v = T::from_protobuf(&mut cursor)?;
             builder.append(Some(v));
         } else {
             builder.append(None);
@@ -105,99 +94,52 @@ fn read_bool_array(array: &PbArray, cardinality: usize) -> ArrayResult<ArrayImpl
     Ok(arr.into())
 }
 
-fn read_date(cursor: &mut Cursor<&[u8]>) -> ArrayResult<Date> {
-    let days = cursor
-        .read_i32::<BigEndian>()
-        .context("failed to read i32 from Date buffer")?;
-
-    Ok(Date::with_days(days)?)
-}
-
-fn read_time(cursor: &mut Cursor<&[u8]>) -> ArrayResult<Time> {
-    let nano = cursor
-        .read_u64::<BigEndian>()
-        .context("failed to read u64 from Time buffer")?;
-
-    Ok(Time::with_nano(nano)?)
-}
-
-fn read_timestamp(cursor: &mut Cursor<&[u8]>) -> ArrayResult<Timestamp> {
-    let micros = cursor
+fn read_offset(offset_cursor: &mut Cursor<&[u8]>) -> ArrayResult<i64> {
+    let offset = offset_cursor
         .read_i64::<BigEndian>()
-        .context("failed to read i64 from Timestamp buffer")?;
+        .context("failed to read i64 from offset buffer")?;
 
-    Ok(Timestamp::with_micros(micros)?)
+    Ok(offset)
 }
 
-fn read_timestamptz(cursor: &mut Cursor<&[u8]>) -> ArrayResult<Timestamptz> {
-    let micros = cursor
-        .read_i64::<BigEndian>()
-        .context("failed to read i64 from Timestamptz buffer")?;
-
-    Timestamptz::from_protobuf(micros)
+trait VarSizedValueReader {
+    type AB: ArrayBuilder;
+    fn new_builder(capacity: usize) -> Self::AB;
+    fn read(buf: &[u8], builder: &mut Self::AB) -> ArrayResult<()>;
 }
 
-fn read_interval(cursor: &mut Cursor<&[u8]>) -> ArrayResult<Interval> {
-    let mut read = || {
-        let months = cursor.read_i32::<BigEndian>()?;
-        let days = cursor.read_i32::<BigEndian>()?;
-        let usecs = cursor.read_i64::<BigEndian>()?;
+struct Utf8ValueReader;
 
-        Ok::<_, std::io::Error>(Interval::from_month_day_usec(months, days, usecs))
-    };
+impl VarSizedValueReader for Utf8ValueReader {
+    type AB = Utf8ArrayBuilder;
 
-    Ok(read().context("failed to read Interval from buffer")?)
-}
-
-macro_rules! read_one_value_array {
-    ($({ $type:ident, $builder:ty }),*) => {
-        paste! {
-            $(
-            fn [<read_ $type:snake _array>](array: &PbArray, cardinality: usize) -> ArrayResult<ArrayImpl> {
-                ensure!(
-                    array.get_values().len() == 1,
-                    "Must have only 1 buffer in a {} array", stringify!($type)
-                );
+    fn new_builder(capacity: usize) -> Self::AB {
+        Utf8ArrayBuilder::new(capacity)
+    }
 
-                let buf = array.get_values()[0].get_body().as_slice();
-
-                let mut builder = $builder::new(cardinality);
-                let bitmap: Bitmap = array.get_null_bitmap()?.into();
-                let mut cursor = Cursor::new(buf);
-                for not_null in bitmap.iter() {
-                    if not_null {
-                        builder.append(Some([<read_ $type:snake>](&mut cursor)?));
-                    } else {
-                        builder.append(None);
-                    }
-                }
-                let arr = builder.finish();
-                ensure_eq!(arr.len(), cardinality);
-
-                Ok(arr.into())
-            }
-            )*
-        }
-    };
+    fn read(buf: &[u8], builder: &mut Utf8ArrayBuilder) -> ArrayResult<()> {
+        let s = std::str::from_utf8(buf).context("failed to read utf8 string from bytes")?;
+        builder.append(Some(s));
+        Ok(())
+    }
 }
 
-read_one_value_array! {
-    { Interval, IntervalArrayBuilder },
-    { Date, DateArrayBuilder },
-    { Time, TimeArrayBuilder },
-    { Timestamp, TimestampArrayBuilder },
-    { Timestamptz, TimestamptzArrayBuilder }
-}
+struct BytesValueReader;
 
-fn read_offset(offset_cursor: &mut Cursor<&[u8]>) -> ArrayResult<i64> {
-    let offset = offset_cursor
-        .read_i64::<BigEndian>()
-        .context("failed to read i64 from offset buffer")?;
+impl VarSizedValueReader for BytesValueReader {
+    type AB = BytesArrayBuilder;
 
-    Ok(offset)
+    fn new_builder(capacity: usize) -> Self::AB {
+        BytesArrayBuilder::new(capacity)
+    }
+
+    fn read(buf: &[u8], builder: &mut BytesArrayBuilder) -> ArrayResult<()> {
+        builder.append(Some(buf));
+        Ok(())
+    }
 }
 
-fn read_string_array<B: ArrayBuilder, R: VarSizedValueReader<B>>(
+fn read_string_array<R: VarSizedValueReader>(
     array: &PbArray,
     cardinality: usize,
 ) -> ArrayResult<ArrayImpl> {
@@ -208,7 +150,7 @@ fn read_string_array<B: ArrayBuilder, R: VarSizedValueReader<B>>(
     let offset_buff = array.get_values()[0].get_body().as_slice();
     let data_buf = array.get_values()[1].get_body().as_slice();
 
-    let mut builder = B::new(cardinality);
+    let mut builder = R::new_builder(cardinality);
     let bitmap: Bitmap = array.get_null_bitmap()?.into();
     let mut offset_cursor = Cursor::new(offset_buff);
     let mut data_cursor = Cursor::new(data_buf);
diff --git a/src/common/src/array/value_reader.rs b/src/common/src/array/value_reader.rs
deleted file mode 100644
index b418a0c315adf..0000000000000
--- a/src/common/src/array/value_reader.rs
+++ /dev/null
@@ -1,88 +0,0 @@
-// Copyright 2024 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::io::Cursor;
-use std::str::from_utf8;
-
-use anyhow::Context;
-use byteorder::{BigEndian, ReadBytesExt};
-
-use super::ArrayResult;
-use crate::array::{
-    ArrayBuilder, BytesArrayBuilder, PrimitiveArrayItemType, Serial, Utf8ArrayBuilder,
-};
-use crate::types::{Decimal, F32, F64};
-
-/// Reads an encoded buffer into a value.
-pub trait PrimitiveValueReader<T: PrimitiveArrayItemType> {
-    fn read(cur: &mut Cursor<&[u8]>) -> ArrayResult<T>;
-}
-
-pub struct I16ValueReader;
-pub struct I32ValueReader;
-pub struct I64ValueReader;
-pub struct SerialValueReader;
-pub struct F32ValueReader;
-pub struct F64ValueReader;
-
-macro_rules! impl_numeric_value_reader {
-    ($value_type:ty, $value_reader:ty, $read_fn:ident) => {
-        impl PrimitiveValueReader<$value_type> for $value_reader {
-            fn read(cur: &mut Cursor<&[u8]>) -> ArrayResult<$value_type> {
-                let v = cur
-                    .$read_fn::<BigEndian>()
-                    .context("failed to read value from buffer")?;
-                Ok(v.into())
-            }
-        }
-    };
-}
-
-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!(F32, F32ValueReader, read_f32);
-impl_numeric_value_reader!(F64, F64ValueReader, read_f64);
-
-pub struct DecimalValueReader;
-
-impl PrimitiveValueReader<Decimal> for DecimalValueReader {
-    fn read(cur: &mut Cursor<&[u8]>) -> ArrayResult<Decimal> {
-        Decimal::from_protobuf(cur)
-    }
-}
-
-pub trait VarSizedValueReader<AB: ArrayBuilder> {
-    fn read(buf: &[u8], builder: &mut AB) -> ArrayResult<()>;
-}
-
-pub struct Utf8ValueReader;
-
-impl VarSizedValueReader<Utf8ArrayBuilder> for Utf8ValueReader {
-    fn read(buf: &[u8], builder: &mut Utf8ArrayBuilder) -> ArrayResult<()> {
-        let s = from_utf8(buf).context("failed to read utf8 string from bytes")?;
-        builder.append(Some(s));
-        Ok(())
-    }
-}
-
-pub struct BytesValueReader;
-
-impl VarSizedValueReader<BytesArrayBuilder> for BytesValueReader {
-    fn read(buf: &[u8], builder: &mut BytesArrayBuilder) -> ArrayResult<()> {
-        builder.append(Some(buf));
-        Ok(())
-    }
-}
diff --git a/src/common/src/types/datetime.rs b/src/common/src/types/datetime.rs
index fac104b3f5aac..1f941eb0f6b19 100644
--- a/src/common/src/types/datetime.rs
+++ b/src/common/src/types/datetime.rs
@@ -17,9 +17,11 @@
 use std::error::Error;
 use std::fmt::Display;
 use std::hash::Hash;
-use std::io::Write;
+use std::io::{Cursor, Write};
 use std::str::FromStr;
 
+use anyhow::Context;
+use byteorder::{BigEndian, ReadBytesExt};
 use bytes::BytesMut;
 use chrono::{
     DateTime, Datelike, Days, Duration, NaiveDate, NaiveDateTime, NaiveTime, Timelike, Weekday,
@@ -450,6 +452,14 @@ impl Date {
             .num_days_from_ce()
     }
 
+    pub fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<Date> {
+        let days = cur
+            .read_i32::<BigEndian>()
+            .context("failed to read i32 from Date buffer")?;
+
+        Ok(Date::with_days(days)?)
+    }
+
     pub fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize> {
         output
             .write(&(self.0.num_days_from_ce()).to_be_bytes())
@@ -484,6 +494,14 @@ impl Time {
         ))
     }
 
+    pub fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<Time> {
+        let nano = cur
+            .read_u64::<BigEndian>()
+            .context("failed to read u64 from Time buffer")?;
+
+        Ok(Time::with_nano(nano)?)
+    }
+
     pub fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize> {
         output
             .write(
@@ -538,6 +556,14 @@ impl Timestamp {
         }))
     }
 
+    pub fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<Timestamp> {
+        let micros = cur
+            .read_i64::<BigEndian>()
+            .context("failed to read i64 from Timestamp buffer")?;
+
+        Ok(Timestamp::with_micros(micros)?)
+    }
+
     /// Although `Timestamp` takes 12 bytes, we drop 4 bytes in protobuf encoding.
     pub fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize> {
         output
diff --git a/src/common/src/types/interval.rs b/src/common/src/types/interval.rs
index d446669d4a949..30845805ede79 100644
--- a/src/common/src/types/interval.rs
+++ b/src/common/src/types/interval.rs
@@ -16,10 +16,11 @@ use std::cmp::Ordering;
 use std::error::Error;
 use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
-use std::io::Write;
+use std::io::{Cursor, Write};
 use std::ops::{Add, Neg, Sub};
 use std::sync::LazyLock;
 
+use anyhow::Context;
 use byteorder::{BigEndian, NetworkEndian, ReadBytesExt, WriteBytesExt};
 use bytes::BytesMut;
 use num_traits::{CheckedAdd, CheckedNeg, CheckedSub, Zero};
@@ -229,6 +230,18 @@ impl Interval {
         secs_from_day_month as i128 * USECS_PER_SEC as i128 + self.usecs as i128
     }
 
+    pub fn from_protobuf(cursor: &mut Cursor<&[u8]>) -> ArrayResult<Interval> {
+        let mut read = || {
+            let months = cursor.read_i32::<BigEndian>()?;
+            let days = cursor.read_i32::<BigEndian>()?;
+            let usecs = cursor.read_i64::<BigEndian>()?;
+
+            Ok::<_, std::io::Error>(Interval::from_month_day_usec(months, days, usecs))
+        };
+
+        Ok(read().context("failed to read Interval from buffer")?)
+    }
+
     pub fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize> {
         output.write_i32::<BigEndian>(self.months)?;
         output.write_i32::<BigEndian>(self.days)?;
diff --git a/src/common/src/types/mod.rs b/src/common/src/types/mod.rs
index ed424b4f4956c..b13df14a93308 100644
--- a/src/common/src/types/mod.rs
+++ b/src/common/src/types/mod.rs
@@ -619,12 +619,12 @@ impl_self_as_scalar_ref! { &str, &[u8], Int256Ref<'_>, JsonbRef<'_>, ListRef<'_>
 macro_rules! for_all_native_types {
     ($macro:ident) => {
         $macro! {
-            { i16, Int16 },
-            { i32, Int32 },
-            { i64, Int64 },
-            { Serial, Serial },
-            { $crate::types::F32, Float32 },
-            { $crate::types::F64, Float64 }
+            { i16, Int16, read_i16 },
+            { i32, Int32, read_i32 },
+            { i64, Int64, read_i64 },
+            { Serial, Serial, read_i64 },
+            { $crate::types::F32, Float32, read_f32 },
+            { $crate::types::F64, Float64, read_f64 }
         }
     };
 }
diff --git a/src/common/src/types/scalar_impl.rs b/src/common/src/types/scalar_impl.rs
index d222473fa6bbb..43742f74c7b51 100644
--- a/src/common/src/types/scalar_impl.rs
+++ b/src/common/src/types/scalar_impl.rs
@@ -27,7 +27,7 @@ pub trait ScalarPartialOrd: Scalar {
 /// Implement `Scalar` and `ScalarRef` for native type.
 /// For `PrimitiveArrayItemType`, clone is trivial, so `T` is both `Scalar` and `ScalarRef`.
 macro_rules! impl_all_native_scalar {
-    ($({ $scalar_type:ty, $variant_name:ident } ),*) => {
+    ($({ $scalar_type:ty, $_variant_name:ident, $_read_fn:ident } ),*) => {
         $(
             impl Scalar for $scalar_type {
                 type ScalarRefType<'a> = Self;
diff --git a/src/common/src/types/timestamptz.rs b/src/common/src/types/timestamptz.rs
index e5e7cb3c71fdf..34a6bd2465271 100644
--- a/src/common/src/types/timestamptz.rs
+++ b/src/common/src/types/timestamptz.rs
@@ -13,9 +13,11 @@
 // limitations under the License.
 
 use std::error::Error;
-use std::io::Write;
+use std::io::{Cursor, Write};
 use std::str::FromStr;
 
+use anyhow::Context;
+use byteorder::{BigEndian, ReadBytesExt};
 use bytes::BytesMut;
 use chrono::{DateTime, Datelike, TimeZone, Utc};
 use chrono_tz::Tz;
@@ -134,8 +136,11 @@ impl Timestamptz {
             .map_err(|_| format!("'{time_zone}' is not a valid timezone"))
     }
 
-    pub fn from_protobuf(timestamp_micros: i64) -> ArrayResult<Self> {
-        Ok(Self(timestamp_micros))
+    pub fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<Timestamptz> {
+        let micros = cur
+            .read_i64::<BigEndian>()
+            .context("failed to read i64 from Timestamptz buffer")?;
+        Ok(Self(micros))
     }
 
     pub fn to_protobuf(self, output: &mut impl Write) -> ArrayResult<usize> {