From ca75da8b7833551b4bc9a6798c848846e77db759 Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Mon, 10 Aug 2026 20:48:29 +0800 Subject: [PATCH] fix(db): scope postgres catalog lookups to the active schema (#13723) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes #11342. ## Problem Upgrading fails on any deployment using a non-`public` postgres schema: ``` [POSTGRES] Fresh install: creating partitioned table file_list db init failed: SqlxError# relation "file_list" already exists ``` The DDL and DML we emit are **unqualified** and resolve through `search_path`, but the catalog / `information_schema` lookups that decide *what* to emit were pinned to a literal `public`. The two disagree: `get_table_relkind` reports "table missing" for a table that plainly exists, we take the fresh-install branch, and `CREATE TABLE` collides with the table already sitting in the user's schema. **The panic is not the worst case.** When a stale same-named table exists in `public` (old deployment, another tenant), the probe returns `'r'` and we take the *regular-to-partitioned migration* branch against the wrong schema's table — renaming its indexes. That matches the index-name conflicts the reporter described. The reporter found 2 sites; there were 14 in total. ## Approach Replace the literal `'public'` with `current_schema()` inline in each catalog query. It evaluates on the very connection that runs the query — the first existing entry of that connection's `search_path`, i.e. exactly where its unqualified DDL/DML lands — so a probe can never disagree with the writes again. No Rust-side state. Two details: - `pg_class` lookups join `pg_namespace` on `nspname = current_schema()` instead of casting to `regnamespace`: the cast parses its input as an identifier and fails on schema names with special characters (e.g. `My-Schema.X`). - `verify_schema_consistency()` at startup checks the read-write, read-only and DDL pools resolve to the same schema. They come from three separately configured DSNs, and `ZO_META_DDL_DSN` is documented as a different **role**, which can carry its own `search_path` via `ALTER ROLE ... SET search_path` — e.g. `cache_indices` reads on `CLIENT` while `create_index` writes on `CLIENT_DDL`, so a divergence would silently relocate the same bug class. Fail loudly instead. ## Sites fixed | File | Functions | |---|---| | `file_list/postgres.rs` | `get_table_relkind`, `check_partition_exists`, `drop_empty_partitions`, `get_varchar_column_width`, `apply_autovacuum_tuning`, `align_identity_sequence` | | `db/postgres.rs` | `cache_indices` (had **no** schema filter at all), `add_column`, `drop_column`, meta `start_dt` probe | | `migration/adapter/postgres.rs` | `list_tables`, `get_columns`, `get_foreign_keys`, `sync_sequences` | | 2 sea-orm migrations | postgres branch had no schema filter, while the mysql branch correctly used `DATABASE()` | The migration-adapter ones matter independently: under a non-public schema `list_tables()` returned **zero** rows, so the cross-database migration tool reported success having migrated nothing. ## Verification Against a real PostgreSQL 18.4 instance with the app configured with `search_path=o2`: - **Fresh install** into schema `o2`: boots clean, all 121 tables land in `o2`. - **Forced upgrade path** (`db_schema_version` lowered) **with a stale regular `public.file_list` planted**: logs `Table file_list already partitioned, ensuring partitions`, upgrade completes, and the stale `public` table is untouched. On main this exact setup either panics (`relation "file_list" already exists`) or migrates the wrong schema's table. - Every rewritten query shape additionally verified in psql under schema `o2test` and under a schema named `My-Schema.X` (stale-table isolation, partition listing, varchar width, identity-sequence alignment, index cache, adapter `list_tables`). `cargo check --bin openobserve`, `cargo fmt --check`, `cargo clippy -p infra --all-targets` all clean; 108 infra postgres unit tests pass. ## Not covered here 1. **Recovery for already-broken databases.** This prevents the failure but does not unstick installs that are mid-migration with renamed indexes. Worth a follow-up (idempotent cleanup, or a documented recovery procedure). 2. **CI coverage.** There is no job exercising a non-`public` `search_path`, which is why these 14 sites accumulated across different changes. A postgres job with `search_path=o2test` would stop this recurring. --- src/infra/src/db/postgres.rs | 47 +++++++++++++++---- src/infra/src/file_list/postgres.rs | 18 ++++--- ...01_add_anomaly_detection_config_columns.rs | 3 +- ..._add_anomaly_detection_config_folder_fk.rs | 3 +- src/migration/adapter/postgres.rs | 8 ++-- 5 files changed, 59 insertions(+), 20 deletions(-) diff --git a/src/infra/src/db/postgres.rs b/src/infra/src/db/postgres.rs index d923418339..6e8c9ed611 100644 --- a/src/infra/src/db/postgres.rs +++ b/src/infra/src/db/postgres.rs @@ -41,6 +41,35 @@ pub static CLIENT_RO: Lazy> = Lazy::new(|| connect(true, false)); pub static CLIENT_DDL: Lazy> = Lazy::new(|| connect(false, true)); static INDICES: OnceCell> = OnceCell::const_new(); +/// The three pools come from separately configured DSNs whose roles can have +/// different `search_path` defaults. All SQL we emit is unqualified, so if the +/// pools resolve to different schemas, fail at startup instead of corrupting data. +pub async fn verify_schema_consistency() -> Result<()> { + let mut schemas = Vec::new(); + for (label, pool) in [ + ("read-write", &*CLIENT), + ("read-only", &*CLIENT_RO), + ("ddl", &*CLIENT_DDL), + ] { + let schema: Option = sqlx::query_scalar("SELECT current_schema()") + .fetch_one(pool) + .await?; + let schema = schema.ok_or_else(|| { + Error::Message(format!( + "postgres search_path of the {label} connection does not resolve to any existing schema" + )) + })?; + schemas.push((label, schema)); + } + if schemas.iter().any(|(_, s)| s != &schemas[0].1) { + return Err(Error::Message(format!( + "postgres schema mismatch, align the search_path of the roles/DSNs involved: {schemas:?}" + ))); + } + log::info!("[POSTGRES] using schema: {}", schemas[0].1); + Ok(()) +} + fn connect(readonly: bool, ddl: bool) -> Pool { let cfg = config::get_config(); @@ -74,7 +103,7 @@ async fn cache_indices() -> HashSet { DB_QUERY_NUMS .with_label_values(&["select", "pg_indexes"]) .inc(); - let sql = r#"SELECT indexname, tablename FROM pg_indexes;"#; + let sql = r#"SELECT indexname, tablename FROM pg_indexes WHERE schemaname = current_schema();"#; let res = sqlx::query_as::<_, (String, String)>(sql) .fetch_all(&client) .await; @@ -674,6 +703,8 @@ impl super::Db for PostgresDb { } pub async fn create_table() -> Result<()> { + verify_schema_consistency().await?; + let pool = CLIENT_DDL.clone(); DB_QUERY_NUMS.with_label_values(&["create", "meta"]).inc(); @@ -698,7 +729,7 @@ CREATE TABLE IF NOT EXISTS meta DB_QUERY_NUMS .with_label_values(&["select", "information_schema.columns"]) .inc(); - let has_start_dt = sqlx::query_scalar::<_,i64>("SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='meta' AND column_name='start_dt';") + let has_start_dt = sqlx::query_scalar::<_,i64>("SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_schema=current_schema() AND table_name='meta' AND column_name='start_dt';") .fetch_one(&pool) .await?; if has_start_dt == 0 { @@ -832,7 +863,7 @@ pub async fn delete_index(idx_name: &str, table: &str) -> Result<()> { pub async fn add_column(table: &str, column: &str, data_type: &str) -> Result<()> { let pool = CLIENT_DDL.clone(); let check_sql = format!( - "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='{table}' AND column_name='{column}';" + "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_schema=current_schema() AND table_name='{table}' AND column_name='{column}';" ); let has_column = sqlx::query_scalar::<_, i64>(&check_sql) .fetch_one(&pool) @@ -860,7 +891,7 @@ pub async fn add_column(table: &str, column: &str, data_type: &str) -> Result<() pub async fn drop_column(table: &str, column: &str) -> Result<()> { let pool = CLIENT_DDL.clone(); let check_sql = format!( - "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='{table}' AND column_name='{column}';" + "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_schema=current_schema() AND table_name='{table}' AND column_name='{column}';" ); let has_column = sqlx::query_scalar::<_, i64>(&check_sql) .fetch_one(&pool) @@ -1077,11 +1108,11 @@ mod tests { let column = "new_column"; let check_sql = format!( - "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='{table}' AND column_name='{column}';" + "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_schema=current_schema() AND table_name='{table}' AND column_name='{column}';" ); assert_eq!( check_sql, - "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='test_table' AND column_name='new_column';" + "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_schema=current_schema() AND table_name='test_table' AND column_name='new_column';" ); } @@ -1106,11 +1137,11 @@ mod tests { let column = "old_column"; let check_sql = format!( - "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='{table}' AND column_name='{column}';" + "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_schema=current_schema() AND table_name='{table}' AND column_name='{column}';" ); assert_eq!( check_sql, - "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='test_table' AND column_name='old_column';" + "SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_schema=current_schema() AND table_name='test_table' AND column_name='old_column';" ); } diff --git a/src/infra/src/file_list/postgres.rs b/src/infra/src/file_list/postgres.rs index 0ed811dad2..ab36fa3b38 100644 --- a/src/infra/src/file_list/postgres.rs +++ b/src/infra/src/file_list/postgres.rs @@ -2176,8 +2176,9 @@ async fn check_partition_exists(pool: &sqlx::Pool, partition_name: &st SELECT EXISTS ( SELECT 1 FROM pg_class c JOIN pg_inherits i ON c.oid = i.inhrelid + JOIN pg_namespace n ON n.oid = c.relnamespace WHERE c.relname = $1 - AND c.relnamespace = 'public'::regnamespace + AND n.nspname = current_schema() ) AS exists "#; let exists: bool = sqlx::query_scalar(sql) @@ -2306,7 +2307,9 @@ async fn precreate_partitions(pool: &sqlx::Pool) -> Result<()> { /// Get the relkind of a table from pg_class. Returns None if table doesn't exist. /// 'r' = regular table, 'p' = partitioned table async fn get_table_relkind(pool: &sqlx::Pool, table: &str) -> Result> { - let sql = "SELECT relkind::text FROM pg_class WHERE relname = $1 AND relnamespace = 'public'::regnamespace"; + let sql = "SELECT c.relkind::text FROM pg_class c \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + WHERE c.relname = $1 AND n.nspname = current_schema()"; let ret: Option = sqlx::query_scalar(sql) .bind(table) .fetch_optional(pool) @@ -2698,7 +2701,9 @@ async fn apply_autovacuum_tuning(pool: &sqlx::Pool) -> Result<()> { for table in &tables { // Check if autovacuum is already tuned let reloptions: Option = sqlx::query_scalar( - "SELECT array_to_string(reloptions, ',') FROM pg_class WHERE relname = $1 AND relnamespace = 'public'::regnamespace", + "SELECT array_to_string(c.reloptions, ',') FROM pg_class c \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + WHERE c.relname = $1 AND n.nspname = current_schema()", ) .bind(table) .fetch_optional(pool) @@ -2737,7 +2742,7 @@ async fn get_varchar_column_width( ) -> Result> { let width: Option = sqlx::query_scalar( "SELECT character_maximum_length FROM information_schema.columns \ - WHERE table_schema = 'public' AND table_name = $1 AND column_name = $2", + WHERE table_schema = current_schema() AND table_name = $1 AND column_name = $2", ) .bind(table) .bind(column) @@ -2919,7 +2924,7 @@ async fn create_future_partitions(conn: &mut PgConnection, table: &str) -> Resul /// is one past the current maximum. async fn align_identity_sequence(conn: &mut PgConnection, table: &str) -> Result<()> { sqlx::query(&format!( - "SELECT setval(pg_get_serial_sequence('public.{table}', 'id'), COALESCE((SELECT MAX(id) FROM public.{table}), 0) + 1, false)" + "SELECT setval(pg_get_serial_sequence('{table}', 'id'), COALESCE((SELECT MAX(id) FROM {table}), 0) + 1, false)" )) .execute(&mut *conn) .await?; @@ -3367,8 +3372,9 @@ SELECT c.relname FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid JOIN pg_class p ON p.oid = i.inhparent +JOIN pg_namespace n ON n.oid = p.relnamespace WHERE p.relname = $1 - AND p.relnamespace = 'public'::regnamespace + AND n.nspname = current_schema() AND c.relname != $2 ORDER BY c.relname "#, diff --git a/src/infra/src/table/migration/m20260317_000001_add_anomaly_detection_config_columns.rs b/src/infra/src/table/migration/m20260317_000001_add_anomaly_detection_config_columns.rs index b79e534f4a..c1f457b050 100644 --- a/src/infra/src/table/migration/m20260317_000001_add_anomaly_detection_config_columns.rs +++ b/src/infra/src/table/migration/m20260317_000001_add_anomaly_detection_config_columns.rs @@ -44,7 +44,8 @@ async fn column_exists(manager: &SchemaManager<'_>, column: &str) -> Result format!( "SELECT COUNT(*) AS cnt FROM information_schema.columns \ - WHERE table_name = 'anomaly_detection_config' AND column_name = '{column}'" + WHERE table_name = 'anomaly_detection_config' AND column_name = '{column}' \ + AND table_schema = current_schema()" ), sea_orm::DbBackend::MySql => format!( "SELECT COUNT(*) AS cnt FROM information_schema.columns \ diff --git a/src/infra/src/table/migration/m20260504_000001_add_anomaly_detection_config_folder_fk.rs b/src/infra/src/table/migration/m20260504_000001_add_anomaly_detection_config_folder_fk.rs index 9bcf45afbd..7ce8ad416a 100644 --- a/src/infra/src/table/migration/m20260504_000001_add_anomaly_detection_config_folder_fk.rs +++ b/src/infra/src/table/migration/m20260504_000001_add_anomaly_detection_config_folder_fk.rs @@ -151,7 +151,8 @@ async fn fk_exists(manager: &SchemaManager<'_>) -> Result { format!( "SELECT COUNT(*) AS cnt FROM information_schema.table_constraints \ WHERE constraint_name = '{FK_NAME}' \ - AND table_name = 'anomaly_detection_config'" + AND table_name = 'anomaly_detection_config' \ + AND table_schema = current_schema()" ), )) .await?; diff --git a/src/migration/adapter/postgres.rs b/src/migration/adapter/postgres.rs index 3f41c6e30a..771a5f4269 100644 --- a/src/migration/adapter/postgres.rs +++ b/src/migration/adapter/postgres.rs @@ -109,7 +109,7 @@ impl DbAdapter for PostgresAdapter { async fn list_tables(&self) -> Result, anyhow::Error> { let rows: Vec<(String,)> = - sqlx::query_as("SELECT tablename FROM pg_tables WHERE schemaname = 'public'") + sqlx::query_as("SELECT tablename FROM pg_tables WHERE schemaname = current_schema()") .fetch_all(&self.pool) .await?; @@ -120,7 +120,7 @@ impl DbAdapter for PostgresAdapter { let rows: Vec<(String, String, String)> = sqlx::query_as( "SELECT column_name, data_type, is_nullable FROM information_schema.columns - WHERE table_schema = 'public' AND table_name = $1 + WHERE table_schema = current_schema() AND table_name = $1 ORDER BY ordinal_position", ) .bind(table) @@ -162,7 +162,7 @@ impl DbAdapter for PostgresAdapter { ON ccu.constraint_name = tc.constraint_name AND ccu.table_schema = tc.table_schema WHERE tc.constraint_type = 'FOREIGN KEY' - AND tc.table_schema = 'public'", + AND tc.table_schema = current_schema()", ) .fetch_all(&self.pool) .await?; @@ -385,7 +385,7 @@ impl DbAdapter for PostgresAdapter { let seq_cols: Vec<(String,)> = sqlx::query_as( "SELECT column_name FROM information_schema.columns - WHERE table_schema = 'public' + WHERE table_schema = current_schema() AND table_name = $1 AND (column_default LIKE 'nextval(%' OR identity_generation IS NOT NULL)",