replicators: Fix DATETIME microsecond precision

This commit fixes the microsecond precision of DATETIME columns in
MySQL by setting the correct fractional seconds precision in the
TimestampTZ object.

This fixes a discrepancy between the precision of the DATETIME in
Readyset and MySQL.

Fixes: REA-4469
Fixes: #1309

Release-Note-Core: Fixes the microsecond precision of DATETIME columns
  in MySQL that sometimes were not being correctly represented in
  Readyset.

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

Author: altmannmarcelo

mysql-srv/src/value/encode.rs

@@ -523,6 +523,7 @@ impl ToMySqlValue for TimestampTz {
                 // 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
+                // TODO(marce): Currently TimestampTZ will produce NULL for zero/invalid dates.
                 let us = ts.nanosecond() / 1_000;
                 let packet_len = if us != 0 {
                     11

readyset-data/src/timestamp.rs

@@ -143,7 +143,7 @@ impl TimestampTz {
 
     /// Set the desired precision when displaying subseconds.
     #[inline(always)]
-    fn set_subsecond_digits(&mut self, count: u8) {
+    pub fn set_subsecond_digits(&mut self, count: u8) {
         self.extra[2] = ((count << TimestampTz::SUBSECOND_DIGITS_BITS.trailing_zeros())
             & TimestampTz::SUBSECOND_DIGITS_BITS)
             | (self.extra[2] & !TimestampTz::SUBSECOND_DIGITS_BITS);

readyset-mysql/tests/integration.rs

@@ -1,7 +1,10 @@
+use std::env;
+use std::panic::AssertUnwindSafe;
 use std::sync::Arc;
 use std::time::Duration;
 
 use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use mysql::Row;
 use mysql_async::prelude::Queryable;
 use mysql_async::OptsBuilder;
 use readyset_adapter::backend::noria_connector::ReadBehavior;
@@ -15,11 +18,13 @@ use readyset_client_test_helpers::{sleep, TestBuilder};
 use readyset_errors::ReadySetError;
 use readyset_server::Handle;
 use readyset_telemetry_reporter::{TelemetryEvent, TelemetryInitializer, TelemetryReporter};
+use readyset_util::eventually;
 use readyset_util::shutdown::ShutdownSender;
 use regex::Regex;
+use serial_test::serial;
 use test_utils::skip_flaky_finder;
 
-async fn setup_with_mysql() -> (mysql_async::Opts, Handle, ShutdownSender) {
+async fn setup_with_mysql(recreate_db: bool) -> (mysql_async::Opts, Handle, ShutdownSender) {
     readyset_tracing::init_test_logging();
     let mut users = std::collections::HashMap::new();
     users.insert("root".to_string(), "noria".to_string());
@@ -30,6 +35,7 @@ async fn setup_with_mysql() -> (mysql_async::Opts, Handle, ShutdownSender) {
             .users(users),
     )
     .fallback(true)
+    .recreate_database(recreate_db)
     .build::<MySQLAdapter>()
     .await
 }
@@ -51,6 +57,14 @@ async fn setup_telemetry() -> (TelemetryReporter, mysql_async::Opts, Handle, Shu
     (reporter, opts, handle, shutdown_tx)
 }
 
+fn mysql_url() -> String {
+    format!(
+        "mysql://root:noria@{}:{}/noria",
+        env::var("MYSQL_HOST").unwrap_or_else(|_| "127.0.0.1".into()),
+        env::var("MYSQL_TCP_PORT").unwrap_or_else(|_| "3306".into()),
+    )
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn duplicate_join_key() {
     // This used to trigger a bug involving weak indexes. See issue #179 for more info.
@@ -1842,8 +1856,9 @@ async fn show_caches_with_always() {
 }
 
 #[tokio::test(flavor = "multi_thread")]
+#[serial]
 async fn show_readyset_status() {
-    let (opts, _handle, shutdown_tx) = setup_with_mysql().await;
+    let (opts, _handle, shutdown_tx) = setup_with_mysql(true).await;
     let mut conn = mysql_async::Conn::new(opts).await.unwrap();
     let mut ret: Vec<mysql::Row> = conn.query("SHOW READYSET STATUS;").await.unwrap();
 
@@ -2107,3 +2122,154 @@ async fn test_proxied_queries_telemetry() {
 
     shutdown_tx.shutdown().await;
 }
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn datetime_nanosecond_precision_text_protocol() {
+    let mut direct_mysql = mysql_async::Conn::from_url(mysql_url()).await.unwrap();
+    direct_mysql.query_drop("DROP TABLE IF EXISTS dt_nano_text_protocol CASCADE;
+             CREATE TABLE dt_nano_text_protocol (col1 DATETIME, col2 DATETIME(2), col3 DATETIME(4), col4 DATETIME(6));
+             INSERT INTO dt_nano_text_protocol VALUES ('2021-01-01 00:00:00', '2021-01-01 00:00:00.00', '2021-01-01 00:00:00.0000', '2021-01-01 00:00:00.000000');
+             INSERT INTO dt_nano_text_protocol VALUES ('2021-01-01 00:00:00', '2021-01-01 00:00:00.01', '2021-01-01 00:00:00.0001', '2021-01-01 00:00:00.000001');")
+        .await
+        .unwrap();
+    let (opts, _handle, shutdown_tx) = setup_with_mysql(false).await;
+    let mut conn = mysql_async::Conn::new(opts).await.unwrap();
+    sleep().await;
+
+    conn.query_drop("CREATE CACHE FROM SELECT * FROM dt_nano_text_protocol")
+        .await
+        .unwrap();
+    sleep().await;
+
+    let my_rows: Vec<(String, String, String, String)> = direct_mysql
+        .query("SELECT * FROM dt_nano_text_protocol")
+        .await
+        .unwrap();
+    let rs_rows: Vec<(String, String, String, String)> = conn
+        .query("SELECT * FROM dt_nano_text_protocol")
+        .await
+        .unwrap();
+    assert_eq!(rs_rows, my_rows);
+    conn.query_drop("INSERT INTO dt_nano_text_protocol VALUES ('2021-01-02 00:00:00', '2021-01-02 00:00:00.00', '2021-01-02 00:00:00.0000', '2021-01-02 00:00:00.000000');").await.unwrap();
+    conn.query_drop("INSERT INTO dt_nano_text_protocol VALUES ('2021-01-02 00:00:00', '2021-01-02 00:00:00.01', '2021-01-02 00:00:00.0001', '2021-01-02 00:00:00.000001');").await.unwrap();
+
+    sleep().await;
+
+    eventually!(run_test: {
+        let my_rows: Vec<(String, String, String, String)> = direct_mysql
+        .query("SELECT * FROM dt_nano_text_protocol")
+        .await
+        .unwrap();
+
+        let rs_rows: Vec<(String, String, String, String)> = conn
+            .query("SELECT * FROM dt_nano_text_protocol")
+            .await
+            .unwrap();
+        AssertUnwindSafe(move || (rs_rows, my_rows))
+    },then_assert: |results| {
+        let (rs_rows, my_rows) = results();
+        assert_eq!(rs_rows, my_rows)
+    });
+
+    shutdown_tx.shutdown().await;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn datetime_nanosecond_precision_binary_protocol() {
+    let mut direct_mysql = mysql_async::Conn::from_url(mysql_url()).await.unwrap();
+    direct_mysql.query_drop("DROP TABLE IF EXISTS dt_nano_bin_protocol CASCADE;
+             CREATE TABLE dt_nano_bin_protocol (ID INT PRIMARY KEY, col1 DATETIME, col2 DATETIME(2), col3 DATETIME(4), col4 DATETIME(6));
+             INSERT INTO dt_nano_bin_protocol VALUES (1, '2021-01-01 00:00:00', '2021-01-01 00:00:00.00', '2021-01-01 00:00:00.0000', '2021-01-01 00:00:00.000000');
+             INSERT INTO dt_nano_bin_protocol VALUES (2, '2021-01-01 00:00:00', '2021-01-01 00:00:00.01', '2021-01-01 00:00:00.0001', '2021-01-01 00:00:00.000001');")
+        .await
+        .unwrap();
+    let (opts, _handle, shutdown_tx) = setup_with_mysql(false).await;
+    let mut conn = mysql_async::Conn::new(opts).await.unwrap();
+    sleep().await;
+
+    conn.query_drop("CREATE CACHE FROM SELECT * FROM dt_nano_bin_protocol WHERE ID = ?")
+        .await
+        .unwrap();
+    sleep().await;
+
+    for id in 1..=2 {
+        let my_rows: Row = direct_mysql
+            .exec_first("SELECT * FROM dt_nano_bin_protocol WHERE ID = ?", (id,))
+            .await
+            .unwrap()
+            .unwrap();
+        let rs_rows: Row = conn
+            .exec_first("SELECT * FROM dt_nano_bin_protocol WHERE ID = ?", (id,))
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(rs_rows.unwrap_raw(), my_rows.unwrap_raw())
+    }
+
+    conn.query_drop("INSERT INTO dt_nano_bin_protocol VALUES (3, '2021-01-02 00:00:00', '2021-01-02 00:00:00.00', '2021-01-02 00:00:00.0000', '2021-01-02 00:00:00.000000');").await.unwrap();
+    conn.query_drop("INSERT INTO dt_nano_bin_protocol VALUES (4, '2021-01-02 00:00:00', '2021-01-02 00:00:00.01', '2021-01-02 00:00:00.0001', '2021-01-02 00:00:00.000001');").await.unwrap();
+
+    sleep().await;
+
+    for id in 3..=4 {
+        eventually!(run_test: {
+
+            let my_rows: Row = direct_mysql
+            .exec_first("SELECT * FROM dt_nano_bin_protocol WHERE ID = ?", (id,))
+            .await
+            .unwrap()
+            .unwrap();
+        let rs_rows: Row = conn
+            .exec_first("SELECT * FROM dt_nano_bin_protocol WHERE ID = ?", (id,))
+            .await
+            .unwrap()
+            .unwrap();
+            AssertUnwindSafe(move || (rs_rows, my_rows))
+        },then_assert: |results| {
+            let (rs_rows, my_rows) = results();
+            assert_eq!(rs_rows.unwrap_raw(), my_rows.unwrap_raw())
+        });
+    }
+    shutdown_tx.shutdown().await;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn datetime_binary_protocol() {
+    let (opts, _handle, shutdown_tx) = setup_with_mysql(false).await;
+    let mut conn = mysql_async::Conn::new(opts).await.unwrap();
+    let mut direct_mysql = mysql_async::Conn::from_url(mysql_url()).await.unwrap();
+    direct_mysql.query_drop("SET sql_mode='';").await.unwrap();
+    direct_mysql
+        .query_drop("DROP TABLE IF EXISTS dt_bin_protocol CASCADE;")
+        .await
+        .unwrap();
+    direct_mysql.query_drop("CREATE TABLE dt_bin_protocol (ID INT PRIMARY KEY, col1 DATETIME(6), col2 DATETIME(6), col3 DATETIME(6), col4 DATETIME(6));").await.unwrap();
+    direct_mysql.query_drop("INSERT INTO dt_bin_protocol VALUES (1, '0000-00-00 00:00:00.000000', '2021-01-01 00:00:00.000000', '2021-01-01 00:00:01.0000000', '2021-01-01 00:00:01.000001');")
+        .await
+        .unwrap();
+    sleep().await;
+
+    conn.query_drop("CREATE CACHE FROM SELECT * FROM dt_bin_protocol WHERE ID = ?")
+        .await
+        .unwrap();
+    sleep().await;
+    let rs_rows: Row = conn
+        .exec_first("SELECT * FROM dt_bin_protocol WHERE ID = 1", ())
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        rs_rows.unwrap_raw(),
+        vec![
+            Some(mysql::Value::Int(1)),
+            Some(mysql::Value::NULL),
+            Some(mysql::Value::Date(2021, 1, 1, 0, 0, 0, 0)),
+            Some(mysql::Value::Date(2021, 1, 1, 0, 0, 1, 0)),
+            Some(mysql::Value::Date(2021, 1, 1, 0, 0, 1, 1))
+        ]
+    );
+    shutdown_tx.shutdown().await;
+}

replicators/src/mysql_connector/connector.rs

@@ -853,18 +853,16 @@ fn binlog_val_to_noria_val(
     // Not all values are coerced to the value expected by ReadySet directly
 
     use mysql_common::constants::ColumnType;
-
-    let buf = match val {
-        mysql_common::value::Value::Bytes(b) => b,
-        _ => {
-            return val.try_into().map_err(|e| {
-                mysql_async::Error::Other(Box::new(internal_err!("Unable to coerce value {}", e)))
-            })
-        }
-    };
-
     match (col_kind, meta) {
         (ColumnType::MYSQL_TYPE_TIMESTAMP2, &[0]) => {
+            let buf = match val {
+                mysql_common::value::Value::Bytes(b) => b,
+                _ => {
+                    return Err(mysql_async::Error::Other(Box::new(internal_err!(
+                        "Expected a byte array for timestamp"
+                    ))));
+                }
+            };
             //https://github.com/blackbeam/rust_mysql_common/blob/408effed435c059d80a9e708bcfa5d974527f476/src/binlog/value.rs#L144
             // When meta is 0, `mysql_common` encodes this value as number of seconds (since UNIX
             // EPOCH)
@@ -881,6 +879,14 @@ fn binlog_val_to_noria_val(
             Ok(time.into())
         }
         (ColumnType::MYSQL_TYPE_TIMESTAMP2, _) => {
+            let buf = match val {
+                mysql_common::value::Value::Bytes(b) => b,
+                _ => {
+                    return Err(mysql_async::Error::Other(Box::new(internal_err!(
+                        "Expected a byte array for timestamp"
+                    ))));
+                }
+            };
             // When meta is anything else, `mysql_common` encodes this value as number of
             // seconds.microseconds (since UNIX EPOCH)
             let s = String::from_utf8_lossy(buf);
@@ -894,6 +900,14 @@ fn binlog_val_to_noria_val(
             Ok(time.into())
         }
         (ColumnType::MYSQL_TYPE_STRING, meta) => {
+            let buf = match val {
+                mysql_common::value::Value::Bytes(b) => b,
+                _ => {
+                    return Err(mysql_async::Error::Other(Box::new(internal_err!(
+                        "Expected a byte array for timestamp"
+                    ))));
+                }
+            };
             match mysql_pad_collation_column(
                 buf,
                 col_kind,
@@ -904,6 +918,28 @@ fn binlog_val_to_noria_val(
                 Err(e) => Err(mysql_async::Error::Other(Box::new(internal_err!("{e}")))),
             }
         }
+        (ColumnType::MYSQL_TYPE_DATETIME2, meta) => {
+            //meta[0] is the fractional seconds precision
+            let df_val: DfValue = val
+                .try_into()
+                .map_err(|e| {
+                    mysql_async::Error::Other(Box::new(internal_err!(
+                        "Unable to coerce value {}",
+                        e
+                    )))
+                })
+                .and_then(|val| match val {
+                    DfValue::TimestampTz(mut ts) => {
+                        ts.set_subsecond_digits(meta[0]);
+                        Ok(DfValue::TimestampTz(ts))
+                    }
+                    DfValue::None => Ok(DfValue::None), //NULL
+                    _ => Err(mysql_async::Error::Other(Box::new(internal_err!(
+                        "Expected a timestamp"
+                    )))),
+                })?;
+            Ok(df_val)
+        }
         _ => Ok(val.try_into().map_err(|e| {
             mysql_async::Error::Other(Box::new(internal_err!("Unable to coerce value {}", e)))
         })?),