refactor(postgres): replace sqlx with tokio-postgres + deadpool-postgres

- Use client.transaction() for safer transaction handling
- Add prepare_cached for all system queries to enable statement caching
- Use query_raw streaming to avoid buffering large result sets
- Add batch_execute support for bulk DDL scripts
- Add COPY protocol support (copy_in / copy_out) for fast data transfer
- Add pg_quote_ident to prevent SQL injection in SET search_path
- Increase connection pool max_size from 5 to 10
- Fix list_indexes bounds check for expression indexes
This commit is contained in:
t8y2 committed 2026-05-24 18:53:20 +08:00
1 parent fd4d3d2338
commit e83c335a70
8 files changed
+774 -757

No files matched your search

+6 -2
View File
@@ -16,10 +16,13 @@ log = "0.4"
tokio = { version = "1", features = ["full"] }
tokio-util = { version = "0.7", features = ["compat"] }
chrono = { version = "0.4", features = ["serde"] }
rust_decimal = { version = "1", features = ["serde"] }
rust_decimal = { version = "1", features = ["serde", "db-postgres"] }
anyhow = "1"
uuid = { version = "1", features = ["v4", "serde"] }
sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "json", "chrono", "uuid", "rust_decimal"] }
tokio-postgres = { version = "0.7", features = ["with-chrono-0_4", "with-uuid-1", "with-serde_json-1"] }
deadpool-postgres = { version = "0.14", features = ["rt_tokio_1"] }
tokio-postgres-rustls = "0.13"
webpki-roots = "0.26"
rusqlite = { version = "0.32", features = ["bundled"] }
mysql_async = { version = "0.36", default-features = false, features = ["default-rustls", "client_ed25519", "chrono", "rust_decimal"] }
sqlparser = "0.62.0"
@@ -37,4 +40,5 @@ csv = "1"
calamine = "0.30.1"
base64 = "0.22"
async-trait = "0.1"
bytes = "1"
zip = { version = "2", default-features = false, features = ["deflate"] }
+1 -1
View File
@@ -41,7 +41,7 @@ pub enum MysqlMode {
pub enum PoolKind {
Mysql(db::mysql::MySqlPool, MysqlMode),
Postgres(sqlx::postgres::PgPool),
Postgres(deadpool_postgres::Pool),
Sqlite(db::sqlite::SqliteHandle),
Redis(tokio::sync::Mutex<redis::aio::MultiplexedConnection>),
DuckDb(Arc<std::sync::Mutex<duckdb::Connection>>),
File diff suppressed because it is too large. Load diff
+12 -12
View File
@@ -916,7 +916,7 @@ pub async fn execute_statements_in_transaction(
/// Owned pool variants for safe dispatch across async boundaries.
enum TxPath {
Pg(sqlx::postgres::PgPool),
Pg(deadpool_postgres::Pool),
Mysql(mysql_async::Pool, bool),
Sqlite(db::sqlite::SqliteHandle),
Explicit,
@@ -925,32 +925,32 @@ enum TxPath {
// Each of these acquires a dedicated connection and runs all statements within
// BEGIN ... COMMIT/ROLLBACK, guaranteeing a single physical connection.
// This avoids sqlx::Transaction<T> which has Send/lifetime incompatibility with Tauri macro.
async fn exec_tx_pg_inner(
pool: sqlx::postgres::PgPool,
pool: deadpool_postgres::Pool,
statements: &[String],
schema: Option<&str>,
start: std::time::Instant,
) -> Result<db::QueryResult, String> {
let mut conn = pool.acquire().await.map_err(|e| format!("Failed to acquire connection: {}", e))?;
// Set schema first
let mut client = pool.get().await.map_err(|e| format!("Failed to acquire connection: {}", e))?;
if let Some(s) = schema {
let sp = format!("SET search_path TO \"{}\", public", s);
sqlx::query(&sp).execute(&mut *conn).await.map_err(|e| format!("SET search_path failed: {}", e))?;
client
.execute(&format!("SET search_path TO {}, public", db::postgres::pg_quote_ident(s)), &[])
.await
.map_err(|e| format!("SET search_path failed: {}", e))?;
}
sqlx::query("BEGIN").execute(&mut *conn).await.map_err(|e| format!("Failed to begin transaction: {}", e))?;
let tx = client.transaction().await.map_err(|e| format!("Failed to begin transaction: {}", e))?;
let mut total_affected: u64 = 0;
for (i, sql) in statements.iter().enumerate() {
match sqlx::query(sql).execute(&mut *conn).await {
Ok(r) => total_affected += r.rows_affected(),
match tx.execute(sql, &[]).await {
Ok(affected) => total_affected += affected,
Err(e) => {
let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
// Transaction auto-rollbacks on drop
return Err(format!("Statement {} failed: {}", i + 1, e));
}
}
}
sqlx::query("COMMIT").execute(&mut *conn).await.map_err(|e| format!("COMMIT failed: {}", e))?;
tx.commit().await.map_err(|e| format!("COMMIT failed: {}", e))?;
Ok(db::QueryResult {
columns: vec![],
rows: vec![],
+1 -1
View File
@@ -1118,7 +1118,7 @@ pub async fn sqlite_ddl(pool: &db::sqlite::SqliteHandle, table: &str) -> Result<
.map_err(|e| e.to_string())?
}
pub async fn pg_ddl(pool: &sqlx::postgres::PgPool, schema: &str, table: &str) -> Result<String, String> {
pub async fn pg_ddl(pool: &deadpool_postgres::Pool, schema: &str, table: &str) -> Result<String, String> {
let (columns, indexes, fkeys) = tokio::try_join!(
db::postgres::get_columns(pool, schema, table),
db::postgres::list_indexes(pool, schema, table),