From 6254f64eb87870ff995d2e5a88725575283c695e Mon Sep 17 00:00:00 2001 From: Lev Kokotov Date: Thu, 23 Jul 2026 14:53:05 -0700 Subject: [PATCH 1/6] feat: schema cache to prevent thundering herd --- pgdog/src/backend/databases.rs | 8 ++++ pgdog/src/backend/pool/shard/mod.rs | 31 ++++++++++--- pgdog/src/backend/schema/cache/mod.rs | 63 +++++++++++++++++++++++++++ pgdog/src/backend/schema/mod.rs | 2 + 4 files changed, 98 insertions(+), 6 deletions(-) create mode 100644 pgdog/src/backend/schema/cache/mod.rs diff --git a/pgdog/src/backend/databases.rs b/pgdog/src/backend/databases.rs index 11d553d94..9a7bdcadb 100644 --- a/pgdog/src/backend/databases.rs +++ b/pgdog/src/backend/databases.rs @@ -19,6 +19,7 @@ use tracing::{debug, error, info, warn}; use crate::auth::AuthResult; use crate::backend::replication::ShardedSchemas; +use crate::backend::schema::SchemaCache; use crate::config::PoolerMode; use crate::frontend::PreparedStatements; use crate::frontend::client::query_engine::two_pc::Manager; @@ -57,6 +58,9 @@ pub fn databases() -> Arc { /// Replace databases pooler-wide. pub fn replace_databases(new_databases: Databases, reload: bool) -> Result<(), Error> { + // Bust the schema cache before launchiung pools. + SchemaCache::global().clear(); + // Order of operations is important // to ensure zero downtime for clients. // @@ -94,6 +98,10 @@ pub fn reload_from_existing() -> Result<(), Error> { let _lock = lock(); let config = config(); let databases = from_config(&config); + + // Bust the schema cache before relaunchiung pools. + SchemaCache::global().clear(); + replace_databases(databases, true)?; Ok(()) } diff --git a/pgdog/src/backend/pool/shard/mod.rs b/pgdog/src/backend/pool/shard/mod.rs index a885fdd6c..953da6694 100644 --- a/pgdog/src/backend/pool/shard/mod.rs +++ b/pgdog/src/backend/pool/shard/mod.rs @@ -14,6 +14,7 @@ use crate::backend::Schema; use crate::backend::databases::User; use crate::backend::pool::lb::ban::Ban; use crate::backend::pub_sub::listener::Listener; +use crate::backend::schema::SchemaCache; use crate::config::{LoadBalancingStrategy, ReadWriteSplit, Role}; use crate::net::Parameters; use crate::net::messages::FrontendPid; @@ -37,7 +38,7 @@ pub(super) struct ShardConfig<'a> { pub(super) lb_strategy: LoadBalancingStrategy, /// Primary/replica read/write split strategy. pub(super) rw_split: ReadWriteSplit, - /// Cluster identifier (user/password). + /// Cluster identifier (user/database). pub(super) identifier: Arc, /// LSN check interval pub(super) lsn_check_interval: Duration, @@ -120,23 +121,41 @@ impl Shard { } } - /// Load schema from the shard's primary. - pub async fn load_schema(&self) -> Result { + /// Load schema from the shard's primary database. + /// + /// Uses the global schema cache, so most requests will not actually touch + /// the database. + /// + pub(crate) async fn load_schema(&self) -> Result { if self.schema.initialized() { return Ok(false); } + // This is syncrhonized by database/shard number, so this prevents + // a thundering herd with 100s of users, for example, all fetching + // the same schema. + let schema = SchemaCache::global().get(self).await?; + let _ = self.schema.set(schema); + self.schema_waiter.notify_one(); + + Ok(true) + } + + /// Fetch schema from the shard. This does not use the + /// cache and returns the freshed schema available. + /// + pub(crate) async fn fetch_schema(&self) -> Result { let mut server = self.primary_or_replica(&Request::default()).await?; let schema = Schema::load(&mut server).await?; + info!( "loaded schema for {} tables on shard {} [{}]", schema.tables().len(), self.number(), server.addr() ); - let _ = self.schema.set(schema); - self.schema_waiter.notify_one(); - Ok(true) + + Ok(schema) } /// Set the schema to its default value. diff --git a/pgdog/src/backend/schema/cache/mod.rs b/pgdog/src/backend/schema/cache/mod.rs new file mode 100644 index 000000000..dc16e6b13 --- /dev/null +++ b/pgdog/src/backend/schema/cache/mod.rs @@ -0,0 +1,63 @@ +//! Cache the schema per database, so we don't have to fetch +//! it for each [`crate::backend::pool::Cluster`]. + +use dashmap::DashMap; +use once_cell::sync::Lazy; +use std::sync::Arc; +use tokio::sync::Mutex; + +use crate::backend::{Schema, Shard}; + +type Entry = Arc>; + +static CACHE: Lazy = Lazy::new(|| SchemaCache::default()); + +/// Schema cache. +#[derive(Debug, Default)] +pub(crate) struct SchemaCache { + // Database => shard => Schema + cache: DashMap>, +} + +impl SchemaCache { + /// Get a schema entry from the cache or load it from + /// the server and store it in the cache. + /// + /// The loading is synchronized with a mutex, so only one user + /// can load a schema at a time, preventing a thundering herd situtation. + /// + pub(crate) async fn get(&self, shard: &Shard) -> Result { + // This is synchronized. + let entry = self + .cache + .entry(shard.identifier().database.clone()) + .or_default() + .entry(shard.number()) + .or_default() + .clone(); + + // This is syncrhonized too, + // so only one shard/user can fetch the schema at a time. + let mut guard = entry.lock().await; + + if guard.is_loaded() { + return Ok(guard.clone()); + } + + let schema = shard.fetch_schema().await?; + + *guard = schema.clone(); + + Ok(schema) + } + + /// Remove all entries from the schema cache. + pub(crate) fn clear(&self) { + self.cache.clear(); + } + + /// Get a reference the the global schema cache. + pub(crate) fn global() -> &'static SchemaCache { + &CACHE + } +} diff --git a/pgdog/src/backend/schema/mod.rs b/pgdog/src/backend/schema/mod.rs index 27d38977d..500c83bc8 100644 --- a/pgdog/src/backend/schema/mod.rs +++ b/pgdog/src/backend/schema/mod.rs @@ -1,4 +1,5 @@ //! Schema operations. +pub mod cache; pub mod columns; pub mod relation; pub mod sync; @@ -11,6 +12,7 @@ use std::ops::DerefMut; use std::{collections::HashMap, ops::Deref}; use tracing::info; +pub(crate) use cache::SchemaCache; pub use relation::Relation; use super::{Cluster, Error, Server, pool::Request}; From b6e153d5ef66bbfa4369ecdb0dc5d5bd452091c5 Mon Sep 17 00:00:00 2001 From: Lev Kokotov Date: Thu, 23 Jul 2026 14:56:22 -0700 Subject: [PATCH 2/6] du --- pgdog/src/backend/databases.rs | 3 --- 1 file changed, 3 deletions(-) diff --git a/pgdog/src/backend/databases.rs b/pgdog/src/backend/databases.rs index 9a7bdcadb..daf795e3b 100644 --- a/pgdog/src/backend/databases.rs +++ b/pgdog/src/backend/databases.rs @@ -99,9 +99,6 @@ pub fn reload_from_existing() -> Result<(), Error> { let config = config(); let databases = from_config(&config); - // Bust the schema cache before relaunchiung pools. - SchemaCache::global().clear(); - replace_databases(databases, true)?; Ok(()) } From e82a2a7f6d3f71f0d0a863230bf91b6a586af7c2 Mon Sep 17 00:00:00 2001 From: Lev Kokotov Date: Thu, 23 Jul 2026 15:28:31 -0700 Subject: [PATCH 3/6] fix test and fmt --- integration/go/go_pgx/pg_tests_test.go | 15 +++++++++++---- integration/go/go_pgx/sharded_test.go | 12 ++++++++---- pgdog/src/backend/schema/cache/mod.rs | 2 +- 3 files changed, 20 insertions(+), 9 deletions(-) diff --git a/integration/go/go_pgx/pg_tests_test.go b/integration/go/go_pgx/pg_tests_test.go index bda5be1d0..265ffc7d9 100644 --- a/integration/go/go_pgx/pg_tests_test.go +++ b/integration/go/go_pgx/pg_tests_test.go @@ -15,8 +15,16 @@ import ( ) func assertShowField(t *testing.T, query string, field string, value int64, user string, database string, shard int64, role string) { - expected_value := value + actualValue := showField(t, query, field, user, database, shard, role) + assert.Equal(t, value, actualValue, fmt.Sprintf("\"%s\" in %s is not %d", field, query, value)) +} + +func assertShowFieldOneOf(t *testing.T, query string, field string, values []int64, user string, database string, shard int64, role string) { + actualValue := showField(t, query, field, user, database, shard, role) + assert.Contains(t, values, actualValue, fmt.Sprintf("\"%s\" in %s is not one of %v", field, query, values)) +} +func showField(t *testing.T, query string, field string, user string, database string, shard int64, role string) int64 { conn, err := pgx.Connect(context.Background(), "postgres://admin:pgdog@127.0.0.1:6432/admin") if err != nil { panic(err) @@ -41,10 +49,9 @@ func assertShowField(t *testing.T, query string, field string, value int64, user if row_db == database && row_user == user && row_shard.Int.Int64() == shard && row_role == role { for i, description := range rows.FieldDescriptions() { if description.Name == field { - actual_value, err := values[i].(pgtype.Numeric).Int64Value() + actualValue, err := values[i].(pgtype.Numeric).Int64Value() assert.NoError(t, err) - assert.Equal(t, expected_value, actual_value.Int64, fmt.Sprintf("\"%s\" in %s is not %d", field, query, value)) - return + return actualValue.Int64 } } } diff --git a/integration/go/go_pgx/sharded_test.go b/integration/go/go_pgx/sharded_test.go index dbc2265bb..abe5b0199 100644 --- a/integration/go/go_pgx/sharded_test.go +++ b/integration/go/go_pgx/sharded_test.go @@ -119,7 +119,9 @@ func rangeCases() []shardCase { } // TestShardedListDeprecated tests list mapping via the legacy [[sharded_mappings]] config format. -func TestShardedListDeprecated(t *testing.T) { runShardingTest(t, "sharded_list_deprecated", listCases()) } +func TestShardedListDeprecated(t *testing.T) { + runShardingTest(t, "sharded_list_deprecated", listCases()) +} // TestShardedList tests list mapping via the inline mapping = [...] config format. func TestShardedList(t *testing.T) { runShardingTest(t, "sharded_list", listCases()) } @@ -228,11 +230,13 @@ func TestShardedTwoPc(t *testing.T) { assert.NoError(t, err) } - // +5 is for schema sync assertShowField(t, "SHOW STATS", "total_xact_2pc_count", 200, "pgdog_2pc", "pgdog_sharded", 0, "primary") assertShowField(t, "SHOW STATS", "total_xact_2pc_count", 200, "pgdog_2pc", "pgdog_sharded", 1, "primary") - assertShowField(t, "SHOW STATS", "total_xact_count", 401+5, "pgdog_2pc", "pgdog_sharded", 0, "primary") // PREPARE, COMMIT for each transaction + TRUNCATE - assertShowField(t, "SHOW STATS", "total_xact_count", 401+5, "pgdog_2pc", "pgdog_sharded", 1, "primary") + // The schema cache is shared across users. If pgdog_2pc wins the cache-fill + // race after RELOAD, the five schema queries are counted in its pool stats. + expectedXactCounts := []int64{401, 406} + assertShowFieldOneOf(t, "SHOW STATS", "total_xact_count", expectedXactCounts, "pgdog_2pc", "pgdog_sharded", 0, "primary") + assertShowFieldOneOf(t, "SHOW STATS", "total_xact_count", expectedXactCounts, "pgdog_2pc", "pgdog_sharded", 1, "primary") for i := range 200 { rows, err := conn.Query( diff --git a/pgdog/src/backend/schema/cache/mod.rs b/pgdog/src/backend/schema/cache/mod.rs index dc16e6b13..dd94621f7 100644 --- a/pgdog/src/backend/schema/cache/mod.rs +++ b/pgdog/src/backend/schema/cache/mod.rs @@ -10,7 +10,7 @@ use crate::backend::{Schema, Shard}; type Entry = Arc>; -static CACHE: Lazy = Lazy::new(|| SchemaCache::default()); +static CACHE: Lazy = Lazy::new(SchemaCache::default); /// Schema cache. #[derive(Debug, Default)] From feed050c5b09d5eefbe7f5ce611514a82dce94c9 Mon Sep 17 00:00:00 2001 From: Lev Kokotov Date: Thu, 23 Jul 2026 15:48:22 -0700 Subject: [PATCH 4/6] remove diff and fix comment --- pgdog/src/backend/databases.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/pgdog/src/backend/databases.rs b/pgdog/src/backend/databases.rs index daf795e3b..31a438164 100644 --- a/pgdog/src/backend/databases.rs +++ b/pgdog/src/backend/databases.rs @@ -58,7 +58,7 @@ pub fn databases() -> Arc { /// Replace databases pooler-wide. pub fn replace_databases(new_databases: Databases, reload: bool) -> Result<(), Error> { - // Bust the schema cache before launchiung pools. + // Bust the schema cache before launching pools. SchemaCache::global().clear(); // Order of operations is important @@ -98,7 +98,6 @@ pub fn reload_from_existing() -> Result<(), Error> { let _lock = lock(); let config = config(); let databases = from_config(&config); - replace_databases(databases, true)?; Ok(()) } From 050ce9054cae3597e9a801aec668382e29060119 Mon Sep 17 00:00:00 2001 From: Lev Kokotov Date: Fri, 24 Jul 2026 10:58:42 -0700 Subject: [PATCH 5/6] Update pgdog/src/backend/schema/cache/mod.rs Co-authored-by: Sage Griffin --- pgdog/src/backend/schema/cache/mod.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/pgdog/src/backend/schema/cache/mod.rs b/pgdog/src/backend/schema/cache/mod.rs index dd94621f7..e8701ee0d 100644 --- a/pgdog/src/backend/schema/cache/mod.rs +++ b/pgdog/src/backend/schema/cache/mod.rs @@ -25,7 +25,6 @@ impl SchemaCache { /// /// The loading is synchronized with a mutex, so only one user /// can load a schema at a time, preventing a thundering herd situtation. - /// pub(crate) async fn get(&self, shard: &Shard) -> Result { // This is synchronized. let entry = self From 2fe3347b3d21cf84ce834c7c7338429612bbf794 Mon Sep 17 00:00:00 2001 From: Lev Kokotov Date: Fri, 24 Jul 2026 12:15:02 -0700 Subject: [PATCH 6/6] scope schema cache to the lifetime of the config --- pgdog/src/backend/databases.rs | 25 +++-- pgdog/src/backend/pool/cluster.rs | 91 +++++++++++-------- pgdog/src/backend/pool/shard/mod.rs | 9 +- pgdog/src/backend/pool/shard/monitor.rs | 1 + pgdog/src/backend/pool/shard/role_detector.rs | 2 + pgdog/src/backend/schema/cache/mod.rs | 17 +--- 6 files changed, 85 insertions(+), 60 deletions(-) diff --git a/pgdog/src/backend/databases.rs b/pgdog/src/backend/databases.rs index 31a438164..5633f00fc 100644 --- a/pgdog/src/backend/databases.rs +++ b/pgdog/src/backend/databases.rs @@ -58,9 +58,6 @@ pub fn databases() -> Arc { /// Replace databases pooler-wide. pub fn replace_databases(new_databases: Databases, reload: bool) -> Result<(), Error> { - // Bust the schema cache before launching pools. - SchemaCache::global().clear(); - // Order of operations is important // to ensure zero downtime for clients. // @@ -557,7 +554,20 @@ fn resolve_table_mapping_deprecated( ) } -fn new_pool(user: &crate::config::User, config: &crate::config::Config) -> Option<(User, Cluster)> { +// Create new Cluster from user and databases in `pgdog.toml`. +// +// # Arguments +// +// - `user`: `[[users]]` entry in `users.toml` +// - `config`: all of `pgdog.toml` +// - `schema_cache`: A cache of database tables, shared between all clusters. This is passed here +// to ensure all clusters share the same schema cache, and to make sure a new one +// is created on each config reload. +fn new_pool( + user: &crate::config::User, + config: &crate::config::Config, + schema_cache: SchemaCache, +) -> Option<(User, Cluster)> { let omnisharded_tables = config.omnisharded_tables(); let sharded_mappings = config.sharded_mappings(); let sharded_schemas = config.sharded_schemas(); @@ -628,6 +638,7 @@ fn new_pool(user: &crate::config::User, config: &crate::config::Config) -> Optio sharded_tables, sharded_schemas, query_parser, + schema_cache, ); Some(( @@ -642,6 +653,8 @@ fn new_pool(user: &crate::config::User, config: &crate::config::Config) -> Optio /// Load databases from config. pub fn from_config(config: &ConfigAndUsers) -> Databases { let mut databases = HashMap::new(); + // The schema cache is shared between all databases. + let schema_cache = SchemaCache::default(); for user in &config.users.users { let users = if user.databases.is_empty() && !user.all_databases { @@ -677,7 +690,7 @@ pub fn from_config(config: &ConfigAndUsers) -> Databases { }; for user in users { - if let Some((user, cluster)) = new_pool(&user, &config.config) { + if let Some((user, cluster)) = new_pool(&user, &config.config, schema_cache.clone()) { databases.insert(user, cluster); } } @@ -1700,7 +1713,7 @@ mod tests { ..Default::default() }; - let result = new_pool(&user, &config); + let result = new_pool(&user, &config, SchemaCache::default()); assert!( result.is_none(), "new_pool should return None when database doesn't exist" diff --git a/pgdog/src/backend/pool/cluster.rs b/pgdog/src/backend/pool/cluster.rs index ff843a61c..9899b1072 100644 --- a/pgdog/src/backend/pool/cluster.rs +++ b/pgdog/src/backend/pool/cluster.rs @@ -8,6 +8,7 @@ use pgdog_config::{ }; use std::{sync::Arc, time::Duration}; +use crate::backend::schema::SchemaCache; use crate::frontend::router::sharding::ShardedTable; use crate::{ backend::{ @@ -126,47 +127,52 @@ impl ClusterShardConfig { /// Cluster creation config. #[derive(Debug)] pub struct ClusterConfig<'a> { - pub name: &'a str, - pub shards: &'a [ClusterShardConfig], - pub lb_strategy: LoadBalancingStrategy, - pub user: &'a str, - pub passwords: Vec, - pub pooler_mode: PoolerMode, - pub sharded_tables: ShardedTables, - pub replication_sharding: Option, - pub multi_tenant: &'a Option, - pub rw_strategy: ReadWriteStrategy, - pub rw_split: ReadWriteSplit, - pub schema_admin: bool, - pub cross_shard_disabled: bool, - pub two_pc: bool, - pub two_pc_auto: bool, - pub sharded_schemas: ShardedSchemas, - pub rewrite: &'a Rewrite, - pub prepared_statements: &'a PreparedStatements, - pub dry_run: bool, - pub expanded_explain: bool, - pub pub_sub_channel_size: usize, - pub query_parser: QueryParserLevel, - pub query_parser_engine: QueryParserEngine, - pub log_min_duration_parse: Option, - pub log_query_sample_length: usize, - pub connection_recovery: ConnectionRecovery, - pub client_connection_recovery: ConnectionRecovery, - pub lsn_check_interval: Duration, - pub reload_schema_on_ddl: bool, - pub load_schema: LoadSchema, - pub resharding_parallel_copies: usize, - pub resharding_copy_retry_max_attempts: usize, - pub resharding_copy_retry_min_delay: u64, - pub resharding_replication_retry_max_attempts: usize, - pub resharding_replication_retry_min_delay: u64, - pub regex_parser_limit: usize, - pub pub_sub_enabled: bool, - pub identity: &'a Option, + name: &'a str, + shards: &'a [ClusterShardConfig], + lb_strategy: LoadBalancingStrategy, + user: &'a str, + passwords: Vec, + pooler_mode: PoolerMode, + sharded_tables: ShardedTables, + replication_sharding: Option, + multi_tenant: &'a Option, + rw_strategy: ReadWriteStrategy, + rw_split: ReadWriteSplit, + schema_admin: bool, + cross_shard_disabled: bool, + two_pc: bool, + two_pc_auto: bool, + sharded_schemas: ShardedSchemas, + rewrite: &'a Rewrite, + prepared_statements: &'a PreparedStatements, + dry_run: bool, + expanded_explain: bool, + pub_sub_channel_size: usize, + query_parser: QueryParserLevel, + query_parser_engine: QueryParserEngine, + log_min_duration_parse: Option, + log_query_sample_length: usize, + connection_recovery: ConnectionRecovery, + client_connection_recovery: ConnectionRecovery, + lsn_check_interval: Duration, + reload_schema_on_ddl: bool, + load_schema: LoadSchema, + resharding_parallel_copies: usize, + resharding_copy_retry_max_attempts: usize, + resharding_copy_retry_min_delay: u64, + resharding_replication_retry_max_attempts: usize, + resharding_replication_retry_min_delay: u64, + regex_parser_limit: usize, + pub_sub_enabled: bool, + identity: &'a Option, + schema_cache: SchemaCache, } impl<'a> ClusterConfig<'a> { + /// Dependencies for creating a Cluster. + /// + /// TODO(lev): This is getting unruly. We may need a struct for a struct :) + /// pub(crate) fn new( config: &'a crate::config::Config, user: &'a User, @@ -174,6 +180,7 @@ impl<'a> ClusterConfig<'a> { sharded_tables: ShardedTables, sharded_schemas: ShardedSchemas, query_parser: QueryParser, + schema_cache: SchemaCache, ) -> Self { let general = &config.general; let multi_tenant = config.multi_tenant(); @@ -228,6 +235,7 @@ impl<'a> ClusterConfig<'a> { regex_parser_limit: general.regex_parser_limit, pub_sub_enabled: general.pub_sub_enabled(), identity: &user.identity, + schema_cache, } } } @@ -274,6 +282,7 @@ impl Cluster { regex_parser_limit, pub_sub_enabled, identity, + schema_cache, } = config; let identifier = Arc::new(DatabaseUser { @@ -296,6 +305,7 @@ impl Cluster { identifier: identifier.clone(), lsn_check_interval, pub_sub_enabled, + schema_cache: schema_cache.clone(), }) }) .collect(), @@ -661,6 +671,7 @@ mod test { ConfigAndUsers, OmnishardedTable, PoolerMode, QueryParserLevel, ShardedSchema, }; + use crate::backend::schema::SchemaCache; use crate::frontend::router::sharding::ShardedTable; use crate::{ backend::{ @@ -707,6 +718,7 @@ mod test { identifier: identifier.clone(), lsn_check_interval: Duration::MAX, pub_sub_enabled: false, + schema_cache: SchemaCache::default(), }) }) .collect::>(); @@ -834,6 +846,7 @@ mod test { identifier: cluster.identifier.clone(), lsn_check_interval: Duration::MAX, pub_sub_enabled: false, + schema_cache: SchemaCache::default(), }); cluster } @@ -857,6 +870,7 @@ mod test { identifier: identifier.clone(), lsn_check_interval: Duration::default(), pub_sub_enabled: false, + schema_cache: SchemaCache::default(), })], prepared_statements: config.config.general.prepared_statements, dry_run: config.config.general.dry_run, @@ -891,6 +905,7 @@ mod test { identifier, lsn_check_interval: Duration::default(), pub_sub_enabled: false, + schema_cache: SchemaCache::default(), }); cluster diff --git a/pgdog/src/backend/pool/shard/mod.rs b/pgdog/src/backend/pool/shard/mod.rs index 953da6694..77a571e4c 100644 --- a/pgdog/src/backend/pool/shard/mod.rs +++ b/pgdog/src/backend/pool/shard/mod.rs @@ -44,6 +44,8 @@ pub(super) struct ShardConfig<'a> { pub(super) lsn_check_interval: Duration, /// Pub/sub enabled pub(super) pub_sub_enabled: bool, + /// Global schema cache. + pub(super) schema_cache: SchemaCache, } /// Connection pools for a single database shard. @@ -134,7 +136,7 @@ impl Shard { // This is syncrhonized by database/shard number, so this prevents // a thundering herd with 100s of users, for example, all fetching // the same schema. - let schema = SchemaCache::global().get(self).await?; + let schema = self.schema_cache.get(self).await?; let _ = self.schema.set(schema); self.schema_waiter.notify_one(); @@ -332,6 +334,7 @@ pub struct ShardInner { schema: Arc>, schema_waiter: Notify, pub_sub_enabled: bool, + schema_cache: SchemaCache, } impl ShardInner { @@ -345,6 +348,7 @@ impl ShardInner { identifier, lsn_check_interval, pub_sub_enabled, + schema_cache, } = shard; let primary = primary.as_ref().map(Pool::new); let lb = LoadBalancer::new(&primary, replicas, lb_strategy, rw_split); @@ -362,6 +366,7 @@ impl ShardInner { schema: Arc::new(OnceCell::new()), schema_waiter: Notify::new(), pub_sub_enabled, + schema_cache, } } } @@ -403,6 +408,7 @@ mod test { }), lsn_check_interval: Duration::MAX, pub_sub_enabled: false, + schema_cache: SchemaCache::default(), }); shard.launch(); @@ -442,6 +448,7 @@ mod test { }), lsn_check_interval: Duration::MAX, pub_sub_enabled: false, + schema_cache: SchemaCache::default(), }); shard.launch(); let mut ids = BTreeSet::new(); diff --git a/pgdog/src/backend/pool/shard/monitor.rs b/pgdog/src/backend/pool/shard/monitor.rs index 586d691fe..38ce85df8 100644 --- a/pgdog/src/backend/pool/shard/monitor.rs +++ b/pgdog/src/backend/pool/shard/monitor.rs @@ -302,6 +302,7 @@ mod test { // wakes on the role-change notification we fire below. lsn_check_interval: Duration::MAX, pub_sub_enabled: false, + schema_cache: SchemaCache::default(), }); // index 0 = replica target, index 1 = primary target. diff --git a/pgdog/src/backend/pool/shard/role_detector.rs b/pgdog/src/backend/pool/shard/role_detector.rs index c6287e004..32e8a43c0 100644 --- a/pgdog/src/backend/pool/shard/role_detector.rs +++ b/pgdog/src/backend/pool/shard/role_detector.rs @@ -46,6 +46,7 @@ mod test { use crate::backend::pool::lsn_monitor::LsnStats; use crate::backend::pool::{Address, Config, PoolConfig}; use crate::backend::replication::publisher::Lsn; + use crate::backend::schema::SchemaCache; use crate::config::{LoadBalancingStrategy, ReadWriteSplit, Role}; use pgdog_stats::LsnStats as StatsLsnStats; @@ -89,6 +90,7 @@ mod test { }), lsn_check_interval: Duration::MAX, pub_sub_enabled: false, + schema_cache: SchemaCache::default(), }) } diff --git a/pgdog/src/backend/schema/cache/mod.rs b/pgdog/src/backend/schema/cache/mod.rs index e8701ee0d..3bfc62295 100644 --- a/pgdog/src/backend/schema/cache/mod.rs +++ b/pgdog/src/backend/schema/cache/mod.rs @@ -2,7 +2,6 @@ //! it for each [`crate::backend::pool::Cluster`]. use dashmap::DashMap; -use once_cell::sync::Lazy; use std::sync::Arc; use tokio::sync::Mutex; @@ -10,13 +9,11 @@ use crate::backend::{Schema, Shard}; type Entry = Arc>; -static CACHE: Lazy = Lazy::new(SchemaCache::default); - /// Schema cache. -#[derive(Debug, Default)] +#[derive(Debug, Default, Clone)] pub(crate) struct SchemaCache { // Database => shard => Schema - cache: DashMap>, + cache: Arc>>, } impl SchemaCache { @@ -49,14 +46,4 @@ impl SchemaCache { Ok(schema) } - - /// Remove all entries from the schema cache. - pub(crate) fn clear(&self) { - self.cache.clear(); - } - - /// Get a reference the the global schema cache. - pub(crate) fn global() -> &'static SchemaCache { - &CACHE - } }