fix comm

src/common/src/types/datetime.rs

@@ -494,23 +494,46 @@ impl Timestamp {
         let secs = cur
             .read_i64::<BigEndian>()
             .context("failed to read i64 from Time buffer")?;
-        let nsecs = cur
-            .read_u32::<BigEndian>()
-            .context("failed to read u32 from Time buffer")?;
-
-        Ok(Timestamp::with_secs_nsecs(secs, nsecs)?)
+        if Self::has_timestamp_namo_format_state(secs) {
+            let secs = Self::remove_timestamp_namo_format_state(secs);
+            let nsecs = cur
+                .read_u32::<BigEndian>()
+                .context("failed to read u32 from Time buffer")?;
+            Ok(Timestamp::with_secs_nsecs(secs, nsecs)?)
+        } else {
+            Ok(Timestamp::with_micros(secs)?)
+        }
     }
 
+    // Since timestamp secs is much smaller than i64, we use the highest 2 bit to store the format information, which is compatible with the old format.
+    // New format: secs(i64) + nsecs(u32)
+    // Old format: micros(i64)
     pub fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize> {
         let timestamp_size = output
-            .write(&(self.0.and_utc().timestamp()).to_be_bytes())
+            .write(
+                &(Self::add_timestamp_namo_format_state(self.0.and_utc().timestamp()))
+                    .to_be_bytes(),
+            )
             .map_err(Into::<ArrayError>::into)?;
         let timestamp_subsec_nanos_size = output
             .write(&(self.0.and_utc().timestamp_subsec_nanos()).to_be_bytes())
             .map_err(Into::<ArrayError>::into)?;
         Ok(timestamp_subsec_nanos_size + timestamp_size)
     }
 
+    fn add_timestamp_namo_format_state(value: i64) -> i64 {
+        value ^ (0b01 << 62)
+    }
+
+    fn has_timestamp_namo_format_state(value: i64) -> bool {
+        let state = (value >> 62) & 0b11;
+        state == 0b10 || state == 0b01
+    }
+
+    fn remove_timestamp_namo_format_state(value: i64) -> i64 {
+        value ^ (0b01 << 62)
+    }
+
     pub fn get_timestamp_nanos(&self) -> i64 {
         self.0.and_utc().timestamp_nanos_opt().unwrap()
     }