feat: query parser backend without protobuf in pg_query

Cargo.toml

@@ -11,6 +11,9 @@ resolver = "2"
 # [patch.crates-io]
 # tokio = { path = "../tokio/tokio" }
 
+# [patch."https://github.com/pgdogdev/pg_query.rs.git"]
+# pg_query = { path = "../pg_query.rs" }
+
 [profile.release]
 codegen-units = 1
 lto = true

integration/pgdog.toml

@@ -20,6 +20,7 @@ two_phase_commit = false
 healthcheck_port = 8080
 tls_certificate = "integration/tls/cert.pem"
 tls_private_key = "integration/tls/key.pem"
+query_parser_engine = "pg_query_raw"
 
 [memory]
 net_buffer = 8096

pgdog-config/src/core.rs

@@ -7,7 +7,7 @@ use tracing::{info, warn};
 use crate::sharding::ShardedSchema;
 use crate::{
     EnumeratedDatabase, Memory, OmnishardedTable, PassthoughAuth, PreparedStatements,
-    QueryParserLevel, ReadWriteSplit, RewriteMode, Role,
+    QueryParserEngine, QueryParserLevel, ReadWriteSplit, RewriteMode, Role,
 };
 
 use super::database::Database;
@@ -426,6 +426,15 @@ impl Config {
             warn!(r#""query_parser_enabled" is deprecated, use "query_parser" = "on" instead"#);
             self.general.query_parser = QueryParserLevel::On;
         }
+
+        if self.general.query_parser_engine == QueryParserEngine::PgQueryRaw {
+            if self.memory.stack_size < 32 * 1024 * 1024 {
+                self.memory.stack_size = 32 * 1024 * 1024;
+                warn!(
+                    r#""pg_query_raw" parser engine requires a large thread stack, setting it to 32MiB for each Tokio worker"#
+                );
+            }
+        }
     }
 
     /// Multi-tenancy is enabled.

pgdog-config/src/general.rs

@@ -5,7 +5,7 @@ use std::path::PathBuf;
 use std::time::Duration;
 
 use crate::pooling::ConnectionRecovery;
-use crate::QueryParserLevel;
+use crate::{QueryParserEngine, QueryParserLevel};
 
 use super::auth::{AuthType, PassthoughAuth};
 use super::database::{LoadBalancingStrategy, ReadWriteSplit, ReadWriteStrategy};
@@ -100,6 +100,9 @@ pub struct General {
     /// Query parser.
     #[serde(default)]
     pub query_parser: QueryParserLevel,
+    /// Query parser engine.
+    #[serde(default)]
+    pub query_parser_engine: QueryParserEngine,
     /// Limit on the number of prepared statements in the server cache.
     #[serde(default = "General::prepared_statements_limit")]
     pub prepared_statements_limit: usize,
@@ -227,6 +230,7 @@ impl Default for General {
             prepared_statements: Self::prepared_statements(),
             query_parser_enabled: Self::query_parser_enabled(),
             query_parser: QueryParserLevel::default(),
+            query_parser_engine: QueryParserEngine::default(),
             prepared_statements_limit: Self::prepared_statements_limit(),
             query_cache_limit: Self::query_cache_limit(),
             passthrough_auth: Self::default_passthrough_auth(),

pgdog-config/src/sharding.rs

@@ -306,3 +306,11 @@ pub enum QueryParserLevel {
     Auto,
     Off,
 }
+
+#[derive(Serialize, Deserialize, Debug, Copy, Clone, PartialEq, Eq, Hash, Default)]
+#[serde(rename_all = "snake_case", deny_unknown_fields)]
+pub enum QueryParserEngine {
+    #[default]
+    PgQueryProtobuf,
+    PgQueryRaw,
+}

pgdog-config/src/util.rs

@@ -1,6 +1,6 @@
 use std::time::Duration;
 
-use rand::{distributions::Alphanumeric, Rng};
+use rand::{distr::Alphanumeric, Rng};
 
 pub fn human_duration_optional(duration: Option<Duration>) -> String {
     if let Some(duration) = duration {
@@ -46,7 +46,7 @@ pub fn human_duration(duration: Duration) -> String {
 
 /// Generate a random string of length n.
 pub fn random_string(n: usize) -> String {
-    rand::thread_rng()
+    rand::rng()
         .sample_iter(&Alphanumeric)
         .take(n)
         .map(char::from)

pgdog-plugin/Cargo.toml

@@ -17,7 +17,7 @@ crate-type = ["rlib", "cdylib"]
 libloading = "0.8"
 libc = "0.2"
 tracing = "0.1"
-pg_query = { git = "https://github.com/pgdogdev/pg_query.rs.git" }
+pg_query = { git = "https://github.com/pgdogdev/pg_query.rs.git", rev = "55cd0ee0adb4f62edfb9c2afe7fbad526de25b0c" }
 pgdog-macros = { path = "../pgdog-macros", version = "0.1.1" }
 toml = "0.9"
 

pgdog/Cargo.toml

@@ -28,7 +28,7 @@ clap = { version = "4", features = ["derive"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 async-trait = "0.1"
-rand = "0.8"
+rand = "0.9.2"
 once_cell = "1"
 tokio-rustls = "0.26"
 rustls-native-certs = "0.8"
@@ -43,7 +43,7 @@ base64 = "0.22"
 md5 = "0.7"
 futures = "0.3"
 csv-core = "0.1"
-pg_query = { git = "https://github.com/pgdogdev/pg_query.rs.git" }
+pg_query = { git = "https://github.com/pgdogdev/pg_query.rs.git", rev = "55cd0ee0adb4f62edfb9c2afe7fbad526de25b0c" }
 regex = "1"
 uuid = { version = "1", features = ["v4", "serde"] }
 url = "2"

pgdog/src/auth/md5.rs

@@ -22,7 +22,7 @@ impl<'a> Client<'a> {
         Self {
             password,
             user,
-            salt: rand::thread_rng().gen(),
+            salt: rand::rng().random(),
         }
     }
 

pgdog/src/auth/scram/server.rs

@@ -52,7 +52,7 @@ impl AuthenticationProvider for UserPassword {
     fn get_password_for(&self, _user: &str) -> Option<PasswordInfo> {
         // TODO: This is slow. We should move it to its own thread pool.
         let iterations = 4096;
-        let salt = rand::thread_rng().gen::<[u8; 16]>().to_vec();
+        let salt = rand::rng().random::<[u8; 16]>().to_vec();
         let hash = hash_password(&self.password, NonZeroU32::new(iterations).unwrap(), &salt);
         Some(PasswordInfo::new(hash.to_vec(), iterations as u16, salt))
     }

pgdog/src/backend/pool/cluster.rs

@@ -1,7 +1,7 @@
 //! A collection of replicas and a primary.
 
 use parking_lot::{Mutex, RwLock};
-use pgdog_config::{PreparedStatements, QueryParserLevel, Rewrite, RewriteMode};
+use pgdog_config::{PreparedStatements, QueryParserEngine, QueryParserLevel, Rewrite, RewriteMode};
 use std::{
     sync::{
         atomic::{AtomicBool, Ordering},
@@ -64,6 +64,7 @@ pub struct Cluster {
     pub_sub_channel_size: usize,
     query_parser: QueryParserLevel,
     connection_recovery: ConnectionRecovery,
+    query_parser_engine: QueryParserEngine,
 }
 
 /// Sharding configuration from the cluster.
@@ -77,6 +78,8 @@ pub struct ShardingSchema {
     pub schemas: ShardedSchemas,
     /// Rewrite config.
     pub rewrite: Rewrite,
+    /// Query parser engine.
+    pub query_parser_engine: QueryParserEngine,
 }
 
 impl ShardingSchema {
@@ -131,6 +134,7 @@ pub struct ClusterConfig<'a> {
     pub expanded_explain: bool,
     pub pub_sub_channel_size: usize,
     pub query_parser: QueryParserLevel,
+    pub query_parser_engine: QueryParserEngine,
     pub connection_recovery: ConnectionRecovery,
     pub lsn_check_interval: Duration,
 }
@@ -177,6 +181,7 @@ impl<'a> ClusterConfig<'a> {
             expanded_explain: general.expanded_explain,
             pub_sub_channel_size: general.pub_sub_channel_size,
             query_parser: general.query_parser,
+            query_parser_engine: general.query_parser_engine,
             connection_recovery: general.connection_recovery,
             lsn_check_interval: Duration::from_millis(general.lsn_check_interval),
         }
@@ -211,6 +216,7 @@ impl Cluster {
             query_parser,
             connection_recovery,
             lsn_check_interval,
+            query_parser_engine,
         } = config;
 
         let identifier = Arc::new(DatabaseUser {
@@ -256,6 +262,7 @@ impl Cluster {
             pub_sub_channel_size,
             query_parser,
             connection_recovery,
+            query_parser_engine,
         }
     }
 
@@ -449,6 +456,7 @@ impl Cluster {
             tables: self.sharded_tables.clone(),
             schemas: self.sharded_schemas.clone(),
             rewrite: self.rewrite.clone(),
+            query_parser_engine: self.query_parser_engine,
         }
     }