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<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';"
         );
     }
 
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<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
             "#,
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<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 \
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<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?;
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<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)",