adapter: Set upstream version in handshake messages

Upon new client connections, Readyset is setting the default database
version string into the connection handshake packets. When we
introduced the `LazyUpstream`, we made some functions not go upstream
and just return a default value; `version()` is one of those
functions.

The reasoning behind the lazy connects is that when control
plane/operational tools connect to Readyset for health
check/diagnostic information, we were inadvertently opening a lot of
connections to the upstream postgres itself, and thus naively
consuming the limited connections that one can make to PG. Also, if a
given client connection will never need to query the upstream anyways,
why create that connection?

This patch adds a new field to the `UpstreamSystemProperties` struct
for `db_version`, and we pass that to the adapter `Backend`, via it's
builder. `Backend::version()` will return that value if it is `Some`;
else, it'll do it's normal behavior (which may be to return the
default).

Fixes: #1562
Release-Note-Core: Set the correct version of the upstream database
  new connection handshake for both mysql and postgres.
Change-Id: Ib4cd2e85a4271c645e90eb9fd44befdfeff202df
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/10703
Tested-by: Buildkite CI
Reviewed-by: Marcelo Altmann <marcelo@readyset.io>

Author: jasobrown-rs

readyset-adapter/src/backend.rs

@@ -314,6 +314,7 @@ pub struct BackendBuilder {
     allow_cache_ddl: bool,
     sampler_tx:
         Option<tokio::sync::mpsc::Sender<(QueryExecutionEvent, String, Vec<SqlIdentifier>)>>,
+    db_version: Option<String>,
 }
 
 impl Default for BackendBuilder {
@@ -337,6 +338,7 @@ impl Default for BackendBuilder {
             connections: None,
             allow_cache_ddl: true,
             sampler_tx: None,
+            db_version: None,
         }
     }
 }
@@ -422,6 +424,7 @@ impl BackendBuilder {
             is_internal_connection: false,
             shallow,
             shallow_refresh_sender,
+            db_version: self.db_version,
             _query_handler: PhantomData,
         }
     }
@@ -525,6 +528,11 @@ impl BackendBuilder {
         self.sampler_tx = tx;
         self
     }
+
+    pub fn db_version(mut self, db_version: String) -> Self {
+        self.db_version = Some(db_version);
+        self
+    }
 }
 
 /// A [`PreparedStatement`] stores the data needed for an immediate execution of a prepared
@@ -652,6 +660,9 @@ where
     /// Sender for shallow refresh requests to worker pool
     shallow_refresh_sender: Option<async_channel::Sender<ShallowRefreshRequest>>,
 
+    /// Memoized upstream database version,
+    db_version: Option<String>,
+
     _query_handler: PhantomData<Handler>,
 }
 
@@ -905,6 +916,10 @@ where
     Handler: 'static + QueryHandler,
 {
     pub fn version(&self) -> String {
+        if let Some(version) = &self.db_version {
+            return version.clone();
+        }
+
         self.upstream
             .as_ref()
             .map(|upstream| upstream.version())

readyset-client-test-helpers/src/lib.rs

@@ -383,6 +383,7 @@ impl TestBuilder {
                                 .lower_case_table_names()
                                 .await
                                 .unwrap(),
+                            db_version: cdc_upstream.version(),
                         }
                     } else {
                         UpstreamSystemProperties {

readyset-data/src/upstream_system_props.rs

@@ -10,6 +10,7 @@ pub struct UpstreamSystemProperties {
     pub timezone_name: SqlIdentifier,
     pub lower_case_database_names: bool,
     pub lower_case_table_names: bool,
+    pub db_version: String,
 }
 
 impl fmt::Display for UpstreamSystemProperties {

readyset/src/lib.rs

@@ -713,12 +713,14 @@ where
         let timezone_name = upstream.timezone_name().await?;
         let lower_case_database_names = upstream.lower_case_database_names().await?;
         let lower_case_table_names = upstream.lower_case_table_names().await?;
+        let db_version = upstream.version();
 
         Ok(UpstreamSystemProperties {
             search_path,
             timezone_name,
             lower_case_database_names,
             lower_case_table_names,
+            db_version,
         })
     };
 
@@ -1408,16 +1410,19 @@ where
                                 .instrument(debug_span!("Building noria connector"))
                                 .await;
 
-                                let backend = backend_builder.clone().build(
-                                    noria,
-                                    upstream,
-                                    Some(upstream_config.clone()),
-                                    query_status_cache,
-                                    adapter_authority.clone(),
-                                    status_reporter_clone,
-                                    adapter_start_time,
-                                    shallow,
-                                );
+                                let backend = backend_builder
+                                    .clone()
+                                    .db_version(sys_props.db_version.clone())
+                                    .build(
+                                        noria,
+                                        upstream,
+                                        Some(upstream_config.clone()),
+                                        query_status_cache,
+                                        adapter_authority.clone(),
+                                        status_reporter_clone,
+                                        adapter_start_time,
+                                        shallow,
+                                    );
                                 connection_handler.process_connection(s, backend).await;
                             }
                             Err(error) => {