mysql_srv: output mysql datetime/timestas as TimestampTZ

We have out own implementation of Datetime/Timestamp called
TimestampTZ. This object has a 3 bytes bitmap that gives us better
control over printing date only, timestamp with tz even if tz is zero
and microsecond precision even if it is zero.
Currently we were converting the TimestampTZ to NaiveDateTime and then
printing it. This was causing some issues with MySQL and datetime.
NaiveDateTime has an inner object called NaiveTime to represent time.
When printing naive time, if the microsecond is zero, it is not
printed [0]. For MySQL, if the Datetime column has the optional
microsecond precision set, we need to print the microsecond even if
it is zero, causing a mismatch between the readyset and MySQL.

This commit changes the display object for datetime/tz columns to
TimestampTZ and implement text and binary protocol trait for it.

Ref: REA-4490
Ref: #1309

[0]: https://github.com/chronotope/chrono/blob/v0.4.38/src/naive/time/mod.rs#L1520

Change-Id: I31301b20bebdd1bb33dbf6b79b84a8f7065dee80
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/7655
Tested-by: Buildkite CI
Reviewed-by: Michael Zink <michael.z@readyset.io>

Author: altmannmarcelo

mysql-srv/src/value/encode.rs

@@ -2,6 +2,7 @@ use std::io::{self, Write};
 
 use byteorder::{LittleEndian, WriteBytesExt};
 use mysql_time::MySqlTime;
+use readyset_data::TimestampTz;
 
 use crate::error::{other_error, OtherErrorKind};
 use crate::myc::constants::{ColumnFlags, ColumnType};
@@ -506,6 +507,70 @@ impl ToMySqlValue for MySqlTime {
     }
 }
 
+impl ToMySqlValue for TimestampTz {
+    fn to_mysql_text<W: Write>(&self, w: &mut W) -> io::Result<()> {
+        w.write_lenenc_str(self.to_string().as_bytes()).map(|_| ())
+    }
+
+    fn to_mysql_bin<W: Write>(&self, w: &mut W, c: &Column) -> io::Result<()> {
+        let ts = self.to_chrono();
+        match c.coltype {
+            ColumnType::MYSQL_TYPE_DATETIME | ColumnType::MYSQL_TYPE_TIMESTAMP => {
+                // https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_binary_resultset.html
+                // To save space the packet can be compressed:
+                // if year, month, day, hour, minutes, seconds and microseconds are all 0, length is
+                // 0 and no other field is sent. if hour, seconds and microseconds
+                // are all 0, length is 4 and no other field is sent.
+                // if microseconds is 0, length is 7 and micro_seconds is not sent.
+                // otherwise the length is 11
+                let us = ts.nanosecond() / 1_000;
+                let packet_len = if us != 0 {
+                    11
+                } else if (ts.hour(), ts.minute(), ts.second()) != (0, 0, 0) {
+                    7
+                } else if (ts.year(), ts.month(), ts.day()) != (0, 0, 0) {
+                    4
+                } else {
+                    0
+                };
+
+                w.write_u8(packet_len)?;
+
+                if packet_len == 0 {
+                    return Ok(()); // no need to write anything else
+                }
+
+                w.write_u16::<LittleEndian>(ts.year() as u16)?;
+                w.write_u8(ts.month() as u8)?;
+                w.write_u8(ts.day() as u8)?;
+
+                if packet_len == 4 {
+                    return Ok(()); // no need to write time
+                }
+
+                w.write_u8(ts.hour() as u8)?;
+                w.write_u8(ts.minute() as u8)?;
+                w.write_u8(ts.second() as u8)?;
+
+                if packet_len == 7 {
+                    return Ok(()); // no need to write microseconds
+                }
+
+                w.write_u32::<LittleEndian>(us)?;
+
+                Ok(())
+            }
+            ColumnType::MYSQL_TYPE_DATE => {
+                if ts.time() != NaiveTime::from_hms_opt(0, 0, 0).unwrap() {
+                    return Err(bad(self, c));
+                }
+                ts.date_naive().to_mysql_bin(w, c)
+            }
+            _ => Err(bad(self, c)),
+        }
+    }
+}
+
 impl ToMySqlValue for NaiveDateTime {
     fn to_mysql_text<W: Write>(&self, w: &mut W) -> io::Result<()> {
         let us = self.nanosecond() / 1_000;

readyset-mysql/src/backend.rs

@@ -139,9 +139,7 @@ async fn write_column<W: AsyncWrite + Unpin>(
             mysql_srv::ColumnType::MYSQL_TYPE_DATETIME
             | mysql_srv::ColumnType::MYSQL_TYPE_DATETIME2
             | mysql_srv::ColumnType::MYSQL_TYPE_TIMESTAMP
-            | mysql_srv::ColumnType::MYSQL_TYPE_TIMESTAMP2 => {
-                rw.write_col(ts.to_chrono().naive_local())
-            }
+            | mysql_srv::ColumnType::MYSQL_TYPE_TIMESTAMP2 => rw.write_col(ts),
             ColumnType::MYSQL_TYPE_DATE => rw.write_col(ts.to_chrono().naive_local().date()),
             _ => return Err(conv_error())?,
         },