test: fix sqlx compatibility and adds integration test for sqlx (#1686)

* test: fix sqlx compatibility and adds integration test for sqlx

* test: correct insert statements

Cargo.lock

@@ -6248,9 +6248,9 @@ dependencies = [
 
 [[package]]
 name = "pgwire"
-version = "0.14.0"
+version = "0.14.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bd66851a4b1d6631371c931810e453b0319eb260bbd5853ebe98e37b15105b80"
+checksum = "bd92c65406efd0d621cdece478a41a89e472a559e44a6f2b218df4c14e66a888"
 dependencies = [
  "async-trait",
  "base64 0.21.2",

src/servers/Cargo.toml

@@ -48,7 +48,7 @@ once_cell = "1.16"
 openmetrics-parser = "0.4"
 opensrv-mysql = "0.4"
 parking_lot = "0.12"
-pgwire = "0.14"
+pgwire = "0.14.1"
 pin-project = "1.0"
 postgres-types = { version = "0.2", features = ["with-chrono-0_4"] }
 promql-parser = "0.1.1"

tests-integration/src/test_util.rs

@@ -511,7 +511,6 @@ pub async fn setup_mysql_server(
     (fe_mysql_addr, guard, fe_mysql_server)
 }
 
-#[allow(dead_code)]
 pub async fn setup_pg_server(
     store_type: StorageType,
     name: &str,

tests-integration/tests/sql.rs

@@ -12,8 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 use sqlx::mysql::MySqlPoolOptions;
+use sqlx::postgres::PgPoolOptions;
 use sqlx::Row;
-use tests_integration::test_util::{setup_mysql_server, StorageType};
+use tests_integration::test_util::{setup_mysql_server, setup_pg_server, StorageType};
 
 #[macro_export]
 macro_rules! sql_test {
@@ -46,6 +47,7 @@ macro_rules! sql_tests {
                 $service,
 
                 test_mysql_crud,
+                test_postgres_crud,
             );
         )*
     };
@@ -65,7 +67,7 @@ pub async fn test_mysql_crud(store_type: StorageType) {
         .await
         .unwrap();
     for i in 0..10 {
-        sqlx::query("insert table demo values(?, ?)")
+        sqlx::query("insert into demo values(?, ?)")
             .bind(i)
             .bind(i)
             .execute(&pool)
@@ -97,3 +99,50 @@ pub async fn test_mysql_crud(store_type: StorageType) {
     let _ = fe_mysql_server.shutdown().await;
     guard.remove_all().await;
 }
+
+pub async fn test_postgres_crud(store_type: StorageType) {
+    let (addr, mut guard, fe_pg_server) = setup_pg_server(store_type, "sql_crud").await;
+
+    let pool = PgPoolOptions::new()
+        .max_connections(2)
+        .connect(&format!("postgres://{addr}/public"))
+        .await
+        .unwrap();
+
+    sqlx::query("create table demo(i bigint, ts timestamp time index)")
+        .execute(&pool)
+        .await
+        .unwrap();
+    for i in 0..10 {
+        sqlx::query("insert into demo values($1, $2)")
+            .bind(i)
+            .bind(i)
+            .execute(&pool)
+            .await
+            .unwrap();
+    }
+
+    let rows = sqlx::query("select i from demo")
+        .fetch_all(&pool)
+        .await
+        .unwrap();
+    assert_eq!(rows.len(), 10);
+
+    for (i, row) in rows.iter().enumerate() {
+        let ret: i64 = row.get(0);
+        assert_eq!(ret, i as i64);
+    }
+
+    sqlx::query("delete from demo")
+        .execute(&pool)
+        .await
+        .unwrap();
+    let rows = sqlx::query("select i from demo")
+        .fetch_all(&pool)
+        .await
+        .unwrap();
+    assert_eq!(rows.len(), 0);
+
+    let _ = fe_pg_server.shutdown().await;
+    guard.remove_all().await;
+}