fix(db): scope postgres catalog lookups to the active schema (#13723)

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

@@ -41,6 +41,35 @@ pub static CLIENT_RO: Lazy<Pool<Postgres>> = Lazy::new(|| connect(true, false));
 pub static CLIENT_DDL: Lazy<Pool<Postgres>> = Lazy::new(|| connect(false, true));
 static INDICES: OnceCell<HashSet<DBIndex>> = 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<String> = 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<Postgres> {
     let cfg = config::get_config();
 
@@ -74,7 +103,7 @@ async fn cache_indices() -> HashSet<DBIndex> {
     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';"
         );
     }
 

src/infra/src/file_list/postgres.rs

@@ -2176,8 +2176,9 @@ async fn check_partition_exists(pool: &sqlx::Pool<Postgres>, 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<Postgres>) -> 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<Postgres>, table: &str) -> Result<Option<String>> {
-    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<String> = sqlx::query_scalar(sql)
         .bind(table)
         .fetch_optional(pool)
@@ -2698,7 +2701,9 @@ async fn apply_autovacuum_tuning(pool: &sqlx::Pool<Postgres>) -> Result<()> {
     for table in &tables {
         // Check if autovacuum is already tuned
         let reloptions: Option<String> = 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<Option<i32>> {
     let width: Option<i32> = 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
             "#,

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<bool
         ),
         sea_orm::DbBackend::Postgres => 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 \

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<bool, DbErr> {
             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?;

src/migration/adapter/postgres.rs

@@ -109,7 +109,7 @@ impl DbAdapter for PostgresAdapter {
 
     async fn list_tables(&self) -> Result<Vec<String>, 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)",