readyset-mysql: Encode results according to @@character_set_results

This is accomplished by plumbing the character set information into the
connector and using it when writing out results. All text/string types
are encoded in the target character set before being written out to the
MySQL protocol connection.

Fixes: REA-5652
Release-Note-Core: Respect `@@character_set_results` when returning
  latin1 and cp850 text columns in MySQL. `SET NAMES` and `\C` also
  work, but do not yet affect collation in lookups.

Change-Id: Ia0eb481f41dbfc0465bcf0d4ea10c6ac6e7d6d2e
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9376
Tested-by: Buildkite CI
Reviewed-by: Marcelo Altmann <marcelo@readyset.io>

Author: mvzink

readyset-adapter/src/backend.rs

@@ -537,7 +537,7 @@ where
     /// Remote socket address of a connected client
     client_addr: SocketAddr,
     /// ReadySet connector used for reads, and writes when no upstream DB is present
-    noria: NoriaConnector,
+    pub noria: NoriaConnector,
     /// Optional connector to the upstream DB. Used for fallback reads and all writes if it exists
     upstream: Option<DB>,
     /// Map from username to password for all users allowed to connect to the db
@@ -2833,6 +2833,7 @@ where
         }
         if let Some(encoding) = set_results_encoding {
             trace!(?encoding, "Setting results_encoding");
+            noria.set_results_encoding(encoding);
         }
 
         Ok(())

readyset-adapter/src/backend/noria_connector.rs

@@ -15,6 +15,7 @@ use readyset_client::{
     ColumnSchema, GraphvizOptions, ReadQuery, ReaderAddress, ReaderHandle, ReadySetHandle,
     SchemaType, Table, TableOperation, View, ViewCreateRequest, ViewQuery,
 };
+use readyset_data::encoding::Encoding;
 use readyset_data::{Collation, DfType, DfValue, Dialect};
 use readyset_errors::{
     internal_err, invariant_eq, table_err, unsupported, unsupported_err, ReadySetError,
@@ -300,6 +301,10 @@ pub struct NoriaConnector {
     /// supports a multi-element schema search path, the concept of "currently connected database"
     /// in MySQL can be thought of as a schema search path that only has one element.
     schema_search_path: Vec<SqlIdentifier>,
+
+    /// The encoding in which to return text results. Corresponds to `character_set_results` in
+    /// MySQL and `SET NAMES` in both MySQL and Postgres.
+    results_encoding: Encoding,
 }
 
 mod request_handler {
@@ -417,6 +422,7 @@ impl NoriaConnector {
             dialect,
             parse_dialect,
             schema_search_path,
+            results_encoding: Encoding::Utf8,
         }
     }
 
@@ -957,6 +963,16 @@ impl NoriaConnector {
         self.schema_search_path.as_ref()
     }
 
+    /// Set the encoding for result sets
+    pub fn set_results_encoding(&mut self, encoding: Encoding) {
+        self.results_encoding = encoding;
+    }
+
+    /// Returns the encoding for result sets
+    pub fn results_encoding(&self) -> Encoding {
+        self.results_encoding
+    }
+
     pub(crate) async fn resnapshot_table(
         &mut self,
         table: &mut Relation,

readyset-e2e-tests/tests/encoding_mysql.rs

@@ -18,6 +18,7 @@ macro_rules! check_rows {
 }
 
 const CHUNK_SIZE: usize = 1000;
+const CHARACTER_SETS: [&str; 4] = ["latin1", "cp850", "utf8mb3", "utf8mb4"];
 
 /// Tests snapshotting replication of a varchar column with the specified character set.
 /// Verifies that the same utf8 encoded version of the data is stored in Readyset.
@@ -142,30 +143,46 @@ where
             }
         );
 
-        // Verify this chunk after updates
-        let mut my_chunk: Vec<(i64, String, Vec<u8>, i32)> = upstream_conn
-            .exec(
-                "SELECT id, hex, text, counter FROM encoding_table WHERE id >= ? AND id <= ?",
-                (first_id, last_id),
-            )
-            .await
-            .unwrap();
-        my_chunk.sort();
+        // Verify this chunk after updates in all supported character sets
+        for character_set in CHARACTER_SETS {
+            upstream_conn
+                .query_drop(format!(
+                    "SET @@session.character_set_results = {character_set}"
+                ))
+                .await
+                .unwrap();
 
-        let mut rs_chunk: Vec<(i64, String, Vec<u8>, i32)> = rs_conn
-            .exec(
-                "SELECT id, hex, text, counter FROM encoding_table WHERE id >= ? AND id <= ?",
-                (first_id, last_id),
-            )
-            .await
-            .unwrap();
-        rs_chunk.sort();
+            let mut my_chunk: Vec<(i64, String, Vec<u8>, i32)> = upstream_conn
+                .exec(
+                    "SELECT id, hex, text, counter FROM encoding_table WHERE id >= ? AND id <= ?",
+                    (first_id, last_id),
+                )
+                .await
+                .unwrap();
+            my_chunk.sort();
+
+            rs_conn
+                .query_drop(format!(
+                    "SET @@session.character_set_results = {character_set}"
+                ))
+                .await
+                .unwrap();
+
+            let mut rs_chunk: Vec<(i64, String, Vec<u8>, i32)> = rs_conn
+                .exec(
+                    "SELECT id, hex, text, counter FROM encoding_table WHERE id >= ? AND id <= ?",
+                    (first_id, last_id),
+                )
+                .await
+                .unwrap();
+            rs_chunk.sort();
 
-        check_rows!(
-                    my_chunk,
-                    rs_chunk,
-                    "mysql (left) differed from readyset (right) after updates for snapshot update chunk {first_id}-{last_id}",
-                );
+            check_rows!(
+                my_chunk,
+                rs_chunk,
+                "mysql (left) differed from readyset (right) after updates for snapshot update chunk {first_id}-{last_id} with character set {character_set}",
+            );
+        }
     }
 
     shutdown_tx.shutdown().await;
@@ -247,27 +264,45 @@ where
             count == chunk.len()
         });
 
-        let my_streaming_rows_chunk: Vec<(i64, String, Vec<u8>)> = upstream_conn
-            .exec(
-                "SELECT id, hex, text FROM encoding_table WHERE id >= ? AND id <= ? ORDER BY id",
-                (first_id, last_id),
-            )
-            .await
-            .unwrap();
+        for character_set in CHARACTER_SETS {
+            upstream_conn
+                .query_drop(format!(
+                    "SET @@session.character_set_results = {character_set}"
+                ))
+                .await
+                .unwrap();
 
-        let rs_streaming_rows_chunk: Vec<(i64, String, Vec<u8>)> = rs_conn
-            .exec(
-                "SELECT id, hex, text FROM encoding_table WHERE id >= ? AND id <= ? ORDER BY id",
-                (first_id, last_id),
-            )
-            .await
-            .unwrap();
+            let mut my_chunk: Vec<(i64, String, Vec<u8>)> = upstream_conn
+                .exec(
+                    "SELECT id, hex, text FROM encoding_table WHERE id >= ? AND id <= ?",
+                    (first_id, last_id),
+                )
+                .await
+                .unwrap();
+            my_chunk.sort();
 
-        check_rows!(
-            my_streaming_rows_chunk,
-            rs_streaming_rows_chunk,
-            "mysql (left) differed from readyset (right) for streaming replication chunk {first_id}-{last_id}"
-        );
+            rs_conn
+                .query_drop(format!(
+                    "SET @@session.character_set_results = {character_set}"
+                ))
+                .await
+                .unwrap();
+
+            let mut rs_chunk: Vec<(i64, String, Vec<u8>)> = rs_conn
+                .exec(
+                    "SELECT id, hex, text FROM encoding_table WHERE id >= ? AND id <= ?",
+                    (first_id, last_id),
+                )
+                .await
+                .unwrap();
+            rs_chunk.sort();
+
+            check_rows!(
+                my_chunk,
+                rs_chunk,
+                "mysql (left) differed from readyset (right) for streaming replication chunk {first_id}-{last_id} with character set {character_set}",
+            );
+        }
     }
 
     shutdown_tx.shutdown().await;

readyset-mysql/src/backend.rs

@@ -21,6 +21,7 @@ use readyset_adapter::backend::{
 };
 use readyset_adapter::upstream_database::LazyUpstream;
 use readyset_adapter_types::{DeallocateId, PreparedStatementType};
+use readyset_data::encoding::Encoding;
 use readyset_data::{DfType, DfValue, DfValueKind};
 use readyset_errors::{internal, ReadySetError};
 use readyset_util::redacted::{RedactedString, Sensitive};
@@ -58,6 +59,7 @@ async fn write_column<S: AsyncRead + AsyncWrite + Unpin>(
     c: &DfValue,
     cs: &mysql_srv::Column,
     ty: &DfType,
+    encoding: Encoding,
 ) -> Result<(), Error> {
     let conv_error = || ReadySetError::DfValueConversionError {
         src_type: format!("{:?}", DfValueKind::from(c)),
@@ -97,7 +99,8 @@ async fn write_column<S: AsyncRead + AsyncWrite + Unpin>(
                 } else {
                     t.as_str()
                 };
-                rw.write_col(s)
+                let b = encoding.encode(s)?;
+                rw.write_col(&*b)
             }
         }
         DfValue::TinyText(ref t) => {
@@ -109,7 +112,8 @@ async fn write_column<S: AsyncRead + AsyncWrite + Unpin>(
                 } else {
                     t.as_str()
                 };
-                rw.write_col(s)
+                let b = encoding.encode(s)?;
+                rw.write_col(&*b)
             }
         }
         ref dt @ (DfValue::Float(..) | DfValue::Double(..)) => match cs.coltype {
@@ -387,6 +391,7 @@ macro_rules! handle_error {
 async fn handle_readyset_result<S>(
     result: noria_connector::QueryResult<'_>,
     writer: QueryResultWriter<'_, S>,
+    results_encoding: Encoding,
 ) -> io::Result<()>
 where
     S: AsyncRead + AsyncWrite + Unpin,
@@ -440,7 +445,7 @@ where
                         None => &DfType::Unknown,
                     };
 
-                    if let Err(e) = write_column(&mut rw, val, c, ty).await {
+                    if let Err(e) = write_column(&mut rw, val, c, ty, results_encoding).await {
                         return handle_column_write_err(e, rw).await;
                     }
                 }
@@ -505,12 +510,15 @@ where
 async fn handle_execute_result<S>(
     result: Result<QueryResult<'_, LazyUpstream<MySqlUpstream>>, Error>,
     writer: QueryResultWriter<'_, S>,
+    results_encoding: Encoding,
 ) -> io::Result<()>
 where
     S: AsyncRead + AsyncWrite + Unpin,
 {
     match result {
-        Ok(QueryResult::Noria(result)) => handle_readyset_result(result, writer).await,
+        Ok(QueryResult::Noria(result)) => {
+            handle_readyset_result(result, writer, results_encoding).await
+        }
         Ok(QueryResult::Upstream(result)) => handle_upstream_result(result, writer).await,
         Ok(QueryResult::UpstreamBufferedInMemory(..)) => handle_error!(
             Error::ReadySet(readyset_errors::unsupported_err!(
@@ -531,13 +539,16 @@ where
 async fn handle_query_result<S>(
     result: Result<QueryResult<'_, LazyUpstream<MySqlUpstream>>, Error>,
     writer: QueryResultWriter<'_, S>,
+    results_encoding: Encoding,
 ) -> QueryResultsResponse
 where
     S: AsyncRead + AsyncWrite + Unpin,
 {
     match result {
         Ok(QueryResult::Parser(command)) => QueryResultsResponse::Command(command),
-        res => QueryResultsResponse::IoResult(handle_execute_result(res, writer).await),
+        res => QueryResultsResponse::IoResult(
+            handle_execute_result(res, writer, results_encoding).await,
+        ),
     }
 }
 
@@ -680,6 +691,8 @@ where
             info!(target: "client_statement", "Execute: {{id: {id}, params: {:?}}}", value_params)
         }
 
+        let results_encoding = self.noria.noria.results_encoding();
+
         match self.execute(id, &value_params, ()).await {
             Ok(QueryResult::Noria(noria_connector::QueryResult::Select { mut rows, schema })) => {
                 let CachedSchema {
@@ -714,15 +727,17 @@ where
                 while let Some(row) = rows.next() {
                     for (c, ty, val) in izip!(mysql_schema.iter(), column_types.iter(), row.iter())
                     {
-                        if let Err(e) = write_column(&mut rw, val, c, ty).await {
+                        if let Err(e) = write_column(&mut rw, val, c, ty, results_encoding).await {
                             return handle_column_write_err(e, rw).await;
                         };
                     }
                     rw.end_row().await?;
                 }
                 rw.finish().await
             }
-            execute_result => handle_execute_result(execute_result, results).await,
+            execute_result => {
+                handle_execute_result(execute_result, results, results_encoding).await
+            }
         }
     }
 
@@ -744,6 +759,8 @@ where
             "charset" => charset.to_string(),
         )
         .increment(1);
+        let encoding = readyset_data::encoding::Encoding::from_mysql_collation_id(charset);
+        self.noria.noria.set_results_encoding(encoding);
         Ok(())
     }
 
@@ -812,8 +829,9 @@ where
             info!(target: "client_statement", "Query: {query}");
         }
 
+        let results_encoding = self.noria.noria.results_encoding();
         let query_result = self.query(query).await;
-        handle_query_result(query_result, results).await
+        handle_query_result(query_result, results, results_encoding).await
     }
 
     fn password_for_username(&self, username: &str) -> Option<Vec<u8>> {