chore: upgrade to sqlx 0.7.1 (influxdata#8266)

There are a bunch of dependencies in `Cargo.lock` that are related to
mysql. These are NOT compiled at all, and are also not part of `cargo
tree`. The reason for the inclusion is a bug in cargo:

rust-lang/cargo#10801

Co-authored-by: kodiakhq[bot] <49736102+kodiakhq[bot]@users.noreply.github.com>

Author: crepererum

Cargo.lock

@@ -17,7 +17,7 @@ once_cell = "1"
 ordered-float = "3"
 schema = { path = "../schema" }
 sha2 = "0.10"
-sqlx = { version = "0.6", features = ["runtime-tokio-rustls", "postgres", "uuid"] }
+sqlx = { version = "0.7.1", features = ["runtime-tokio-rustls", "postgres", "uuid"] }
 thiserror = "1.0.43"
 uuid = { version = "1", features = ["v4"] }
 workspace-hack = { version = "0.1", path = "../workspace-hack" }

data_types/Cargo.toml

@@ -4,7 +4,6 @@ use super::TableId;
 use generated_types::influxdata::iox::schema::v1 as proto;
 use influxdb_line_protocol::FieldValue;
 use schema::{builder::SchemaBuilder, InfluxColumnType, InfluxFieldType, Schema};
-use sqlx::postgres::PgHasArrayType;
 use std::{
     collections::{BTreeMap, BTreeSet, HashMap},
     convert::TryFrom,
@@ -26,12 +25,6 @@ impl ColumnId {
     }
 }
 
-impl PgHasArrayType for ColumnId {
-    fn array_type_info() -> sqlx::postgres::PgTypeInfo {
-        <i64 as PgHasArrayType>::array_type_info()
-    }
-}
-
 /// Column definitions for a table indexed by their name
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct ColumnsByName(BTreeMap<String, ColumnSchema>);
@@ -328,7 +321,7 @@ impl TryFrom<proto::column_schema::ColumnType> for ColumnType {
 
 /// Set of columns.
 #[derive(Debug, Clone, PartialEq, Eq, sqlx::Type)]
-#[sqlx(transparent)]
+#[sqlx(transparent, no_pg_array)]
 pub struct ColumnSet(Vec<ColumnId>);
 
 impl ColumnSet {

data_types/src/columns.rs

@@ -244,7 +244,7 @@ pub static PARTITION_BY_DAY_PROTO: Lazy<Arc<proto::PartitionTemplate>> = Lazy::n
 
 /// A partition template specified by a namespace record.
 #[derive(Debug, PartialEq, Clone, Default, sqlx::Type)]
-#[sqlx(transparent)]
+#[sqlx(transparent, no_pg_array)]
 pub struct NamespacePartitionTemplateOverride(Option<serialization::Wrapper>);
 
 impl TryFrom<proto::PartitionTemplate> for NamespacePartitionTemplateOverride {
@@ -259,7 +259,7 @@ impl TryFrom<proto::PartitionTemplate> for NamespacePartitionTemplateOverride {
 
 /// A partition template specified by a table record.
 #[derive(Debug, PartialEq, Eq, Clone, Default, sqlx::Type)]
-#[sqlx(transparent)]
+#[sqlx(transparent, no_pg_array)]
 pub struct TablePartitionTemplateOverride(Option<serialization::Wrapper>);
 
 impl TablePartitionTemplateOverride {

data_types/src/partition_template.rs

@@ -7,14 +7,6 @@ yanked = "deny"
 unmaintained = "warn"
 notice = "warn"
 ignore = [
-    # "It was sometimes possible for SQLite versions >= 1.0.12, < 3.39.2 to allow an array-bounds overflow when large
-    # string were input into SQLite's printf function."
-    #
-    # We are not using `printf` with untrusted inputs.
-    #
-    # This is currently blocked by upstream:
-    # https://github.com/launchbadge/sqlx/issues/2346
-    "RUSTSEC-2022-0090",
 ]
 git-fetch-with-cli = true
 

deny.toml

@@ -32,5 +32,5 @@ metric = { path = "../metric" }
 once_cell = { version = "1.18", features = ["parking_lot"] }
 parquet_file = { path = "../parquet_file" }
 tempfile = "3"
-sqlx = { version = "0.6", features = [ "runtime-tokio-rustls" ] }
+sqlx = { version = "0.7.1", features = [ "runtime-tokio-rustls" ] }
 

garbage_collector/Cargo.toml

@@ -18,7 +18,7 @@ parking_lot = { version = "0.12" }
 serde = { version = "1.0", features = ["derive"] }
 siphasher = "0.3"
 snafu = "0.7"
-sqlx = { version = "0.6", features = [ "runtime-tokio-rustls" , "postgres", "uuid", "sqlite" ] }
+sqlx = { version = "0.7.1", features = [ "runtime-tokio-rustls" , "postgres", "uuid", "sqlite" ] }
 sqlx-hotswap-pool = { path = "../sqlx-hotswap-pool" }
 thiserror = "1.0.43"
 tokio = { version = "1.29", features = ["io-util", "macros", "parking_lot", "rt-multi-thread", "time"] }

iox_catalog/Cargo.toml

@@ -329,9 +329,9 @@ async fn new_raw_pool(
     parsed_dsn: &str,
 ) -> Result<sqlx::Pool<Postgres>, sqlx::Error> {
     // sqlx exposes some options as pool options, while other options are available as connection options.
-    let mut connect_options = PgConnectOptions::from_str(parsed_dsn)?;
-    // the default is INFO, which is frankly surprising.
-    connect_options.log_statements(log::LevelFilter::Trace);
+    let connect_options = PgConnectOptions::from_str(parsed_dsn)?
+        // the default is INFO, which is frankly surprising.
+        .log_statements(log::LevelFilter::Trace);
 
     let app_name = options.app_name.clone();
     let app_name2 = options.app_name.clone(); // just to log below
@@ -816,7 +816,7 @@ RETURNING *;
         .bind(name) // $1
         .bind(partition_template) // $2
         .bind(namespace_id) // $3
-        .fetch_one(&mut tx)
+        .fetch_one(&mut *tx)
         .await
         .map_err(|e| match e {
             sqlx::Error::RowNotFound => Error::TableCreateLimitError {
@@ -843,7 +843,8 @@ RETURNING *;
         // columns with an unsupported type.
         for template_part in table.partition_template.parts() {
             if let TemplatePart::TagValue(tag_name) = template_part {
-                insert_column_with_connection(&mut tx, tag_name, table.id, ColumnType::Tag).await?;
+                insert_column_with_connection(&mut *tx, tag_name, table.id, ColumnType::Tag)
+                    .await?;
             }
         }
 
@@ -1538,15 +1539,14 @@ WHERE object_store_id = $1;
     ) -> Result<Vec<Uuid>> {
         sqlx::query(
             // sqlx's readme suggests using PG's ANY operator instead of IN; see link below.
+            // https://github.com/launchbadge/sqlx/blob/main/FAQ.md#how-can-i-do-a-select--where-foo-in--query
             r#"
 SELECT object_store_id
 FROM parquet_file
 WHERE object_store_id = ANY($1);
              "#,
         )
-        // from https://github.com/launchbadge/sqlx/blob/main/FAQ.md#how-can-i-do-a-select--where-foo-in--query
-        // a bug of the parameter typechecking code requires all array parameters to be slices
-        .bind(&object_store_ids[..]) // $1
+        .bind(object_store_ids) // $1
         .map(|pgr| pgr.get::<Uuid, _>("object_store_id"))
         .fetch_all(&mut self.inner)
         .await
@@ -1576,13 +1576,13 @@ WHERE object_store_id = ANY($1);
             .map_err(|e| Error::StartTransaction { source: e })?;
 
         let marked_at = Timestamp::from(self.time_provider.now());
-        flag_for_delete(&mut tx, delete, marked_at).await?;
+        flag_for_delete(&mut *tx, delete, marked_at).await?;
 
-        update_compaction_level(&mut tx, upgrade, target_level).await?;
+        update_compaction_level(&mut *tx, upgrade, target_level).await?;
 
         let mut ids = Vec::with_capacity(create.len());
         for file in create {
-            let id = create_parquet_file(&mut tx, file).await?;
+            let id = create_parquet_file(&mut *tx, file).await?;
             ids.push(id);
         }
 
@@ -1667,12 +1667,9 @@ async fn flag_for_delete<'q, E>(
 where
     E: Executor<'q, Database = Postgres>,
 {
-    // If I try to do `.bind(parquet_file_ids)` directly, I get a compile error from sqlx.
-    // See https://github.com/launchbadge/sqlx/issues/1744
-    let ids: Vec<_> = ids.iter().map(|p| p.get()).collect();
     let query = sqlx::query(r#"UPDATE parquet_file SET to_delete = $1 WHERE id = ANY($2);"#)
         .bind(marked_at) // $1
-        .bind(&ids[..]); // $2
+        .bind(ids); // $2
     query
         .execute(executor)
         .await
@@ -1689,9 +1686,6 @@ async fn update_compaction_level<'q, E>(
 where
     E: Executor<'q, Database = Postgres>,
 {
-    // If I try to do `.bind(parquet_file_ids)` directly, I get a compile error from sqlx.
-    // See https://github.com/launchbadge/sqlx/issues/1744
-    let ids: Vec<_> = parquet_file_ids.iter().map(|p| p.get()).collect();
     let query = sqlx::query(
         r#"
 UPDATE parquet_file
@@ -1700,7 +1694,7 @@ WHERE id = ANY($2);
         "#,
     )
     .bind(compaction_level) // $1
-    .bind(&ids[..]); // $2
+    .bind(parquet_file_ids); // $2
     query
         .execute(executor)
         .await

iox_catalog/src/postgres.rs

@@ -577,7 +577,7 @@ RETURNING *;
         .bind(name) // $1
         .bind(partition_template) // $2
         .bind(namespace_id) // $3
-        .fetch_one(&mut tx)
+        .fetch_one(&mut *tx)
         .await
         .map_err(|e| match e {
             sqlx::Error::RowNotFound => Error::TableCreateLimitError {
@@ -604,7 +604,8 @@ RETURNING *;
         // columns with an unsupported type.
         for template_part in table.partition_template.parts() {
             if let TemplatePart::TagValue(tag_name) = template_part {
-                insert_column_with_connection(&mut tx, tag_name, table.id, ColumnType::Tag).await?;
+                insert_column_with_connection(&mut *tx, tag_name, table.id, ColumnType::Tag)
+                    .await?;
             }
         }
 
@@ -1451,14 +1452,14 @@ WHERE object_store_id IN ({v});",
 
         for id in delete {
             let marked_at = Timestamp::from(self.time_provider.now());
-            flag_for_delete(&mut tx, *id, marked_at).await?;
+            flag_for_delete(&mut *tx, *id, marked_at).await?;
         }
 
-        update_compaction_level(&mut tx, upgrade, target_level).await?;
+        update_compaction_level(&mut *tx, upgrade, target_level).await?;
 
         let mut ids = Vec::with_capacity(create.len());
         for file in create {
-            let res = create_parquet_file(&mut tx, file.clone()).await?;
+            let res = create_parquet_file(&mut *tx, file.clone()).await?;
             ids.push(res.id);
         }
         tx.commit()
@@ -1562,8 +1563,7 @@ async fn update_compaction_level<'q, E>(
 where
     E: Executor<'q, Database = Sqlite>,
 {
-    // If I try to do `.bind(parquet_file_ids)` directly, I get a compile error from sqlx.
-    // See https://github.com/launchbadge/sqlx/issues/1744
+    // We use a JSON-based "IS IN" check.
     let ids: Vec<_> = parquet_file_ids.iter().map(|p| p.get()).collect();
     let query = sqlx::query(
         r#"

iox_catalog/src/sqlite.rs

@@ -10,7 +10,7 @@ license.workspace = true
 publish = false
 
 [dependencies]
-sqlx = { version = "0.6.3", features = ["runtime-tokio-rustls", "postgres", "json", "tls"] }
+sqlx = { version = "0.7.1", features = ["runtime-tokio-rustls", "postgres", "json", "tls-rustls"] }
 either = "1.8.1"
 futures = "0.3"
 workspace-hack = { version = "0.1", path = "../workspace-hack" }

sqlx-hotswap-pool/Cargo.toml

@@ -31,7 +31,7 @@ rand = "0.8.3"
 regex = "1.9"
 reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls"] }
 snafu = "0.7"
-sqlx = { version = "0.6", features = [ "runtime-tokio-rustls" , "postgres", "uuid" ] }
+sqlx = { version = "0.7.1", features = [ "runtime-tokio-rustls" , "postgres", "uuid" ] }
 tempfile = "3.6.0"
 test_helpers = { path = "../test_helpers", features = ["future_timeout"] }
 tokio = { version = "1.29", features = ["macros", "net", "parking_lot", "rt-multi-thread", "signal", "sync", "time"] }