save

Author: levkk

integration/pgdog.toml

@@ -22,7 +22,8 @@ tls_certificate = "integration/tls/cert.pem"
 tls_private_key = "integration/tls/key.pem"
 query_parser_engine = "pg_query_raw"
 system_catalogs = "omnisharded_sticky"
-reload_schema_on_ddl = false
+reload_schema_on_ddl = true
+cross_shard_backend = "fdw"
 
 [memory]
 net_buffer = 8096

integration/postgres_fdw/pgdog.toml

@@ -1,5 +1,6 @@
-[fdw]
-enabled = true
+
+[general]
+cross_shard_backend = "fdw"
 
 [[databases]]
 name = "pgdog"

pgdog/src/admin/set.rs

@@ -160,6 +160,10 @@ impl Command for Set {
                 config.config.general.client_idle_in_transaction_timeout = self.value.parse()?;
             }
 
+            "cross_shard_backend" => {
+                config.config.general.cross_shard_backend = Self::from_json(&self.value)?;
+            }
+
             _ => return Err(Error::Syntax),
         }
 

pgdog/src/backend/databases.rs

@@ -83,6 +83,12 @@ pub fn reload_from_existing() -> Result<(), Error> {
     let databases = from_config(&config);
 
     replace_databases(databases, true)?;
+
+    // Reconfigure FDW with new schema.
+    if config.config.general.cross_shard_backend.need_fdw() {
+        PostgresLauncher::get().reconfigure();
+    }
+
     Ok(())
 }
 
@@ -122,7 +128,24 @@ pub fn reload() -> Result<(), Error> {
     tls::reload()?;
 
     // Reconfigure FDW with new schema.
-    PostgresLauncher::get().reconfigure();
+    match (
+        old_config.config.general.cross_shard_backend.need_fdw(),
+        new_config.config.general.cross_shard_backend.need_fdw(),
+    ) {
+        (true, true) => {
+            PostgresLauncher::get().reconfigure();
+        }
+
+        (false, true) => {
+            PostgresLauncher::get().launch();
+        }
+
+        (true, false) => {
+            PostgresLauncher::get().shutdown();
+        }
+
+        (false, false) => {}
+    }
 
     // Remove any unused prepared statements.
     PreparedStatements::global()

pgdog/src/backend/fdw/launcher.rs

@@ -82,6 +82,10 @@ impl PostgresLauncher {
         let _ = self.events.send(LauncherEvent::Start);
     }
 
+    pub(crate) fn shutdown(&self) {
+        let _ = self.events.send(LauncherEvent::Shutdown);
+    }
+
     /// Request reconfiguration.
     pub(crate) fn reconfigure(&self) {
         let _ = self.events.send(LauncherEvent::Reconfigure);

pgdog/src/backend/fdw/postgres.rs

@@ -122,8 +122,41 @@ impl PostgresProcess {
         })
     }
 
+    /// Kill any existing process listening on the given port.
+    /// This handles orphaned postgres processes from previous crashes.
+    #[cfg(unix)]
+    async fn kill_existing_on_port(port: u16) {
+        // Use fuser to find and kill any process on the port
+        let result = Command::new("fuser")
+            .arg("-k")
+            .arg(format!("{}/tcp", port))
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .status()
+            .await;
+
+        if let Ok(status) = result {
+            if status.success() {
+                warn!(
+                    "[fdw] killed orphaned process on port {} from previous run",
+                    port
+                );
+                // Give it a moment to fully terminate
+                sleep(Duration::from_millis(100)).await;
+            }
+        }
+    }
+
+    #[cfg(not(unix))]
+    async fn kill_existing_on_port(_port: u16) {
+        // Not implemented for non-unix platforms
+    }
+
     /// Setup and launch Postgres process.
     pub(crate) async fn launch(&mut self) -> Result<(), Error> {
+        // Clean up any orphaned postgres from previous crashes
+        Self::kill_existing_on_port(self.port).await;
+
         info!(
             "[fdw] initializing \"{}\" (PostgreSQL {})",
             self.initdb_dir.display(),
@@ -168,6 +201,22 @@ impl PostgresProcess {
         #[cfg(unix)]
         cmd.process_group(0); // Prevent sigint from terminal.
 
+        // SAFETY: prctl(PR_SET_PDEATHSIG) is async-signal-safe and doesn't
+        // access any shared state. It tells the kernel to send SIGKILL to
+        // this process when its parent dies, preventing orphaned processes.
+        #[cfg(target_os = "linux")]
+        {
+            #[allow(unused_imports)]
+            use std::os::unix::process::CommandExt;
+            unsafe {
+                cmd.pre_exec(|| {
+                    const PR_SET_PDEATHSIG: nix::libc::c_int = 1;
+                    nix::libc::prctl(PR_SET_PDEATHSIG, nix::libc::SIGKILL);
+                    Ok(())
+                });
+            }
+        }
+
         let child = cmd.spawn()?;
 
         self.pid = child.id().map(|pid| pid as i32);
@@ -212,8 +261,22 @@ impl PostgresProcess {
                         }
                     }
 
-                    _ = process.child.wait() => {
-                        error!("[fdw] postgres shut down unexpectedly");
+                    exit_status = process.child.wait() => {
+                        // Drain remaining stderr before reporting shutdown
+                        loop {
+                            let mut remaining = String::new();
+                            match reader.read_line(&mut remaining).await {
+                                Ok(0) => break, // EOF
+                                Ok(_) => {
+                                    if !remaining.is_empty() {
+                                        let remaining = LOG_PREFIX.replace(&remaining, "");
+                                        info!("[fdw::subprocess] {}", remaining.trim());
+                                    }
+                                }
+                                Err(_) => break,
+                            }
+                        }
+                        error!("[fdw] postgres shut down unexpectedly, exit status: {:?}", exit_status);
                         break;
                     }
 

pgdog/src/backend/pool/cluster.rs

@@ -291,7 +291,7 @@ impl Cluster {
             cross_shard_backend,
         };
 
-        if cross_shard_backend.need_fdw() {
+        if cross_shard_backend.need_fdw() && cluster.shards().len() > 1 {
             cluster.fdw_lb = FdwLoadBalancer::new(&cluster).ok();
         }
 

pgdog/src/backend/schema/postgres_fdw/mod.rs

@@ -10,6 +10,9 @@ pub use custom_types::{CustomType, CustomTypeKind, CustomTypes, CUSTOM_TYPES_QUE
 pub use error::Error;
 pub use extensions::{Extension, Extensions, EXTENSIONS_QUERY};
 pub use schema::{FdwServerDef, ForeignTableColumn, ForeignTableSchema, FOREIGN_TABLE_SCHEMA};
-pub use statement::{create_foreign_table, ForeignTableBuilder, PartitionStrategy};
+pub use statement::{
+    create_foreign_table, CreateForeignTableResult, ForeignTableBuilder, PartitionStrategy,
+    TypeMismatch,
+};
 
 pub(crate) use statement::quote_identifier;

pgdog/src/backend/schema/postgres_fdw/postgres_fdw.sql

@@ -27,6 +27,7 @@ WHERE
     AND n.nspname <> 'pg_catalog'
     AND n.nspname !~ '^pg_toast'
     AND n.nspname <> 'information_schema'
+    AND NOT (n.nspname = 'pgdog' AND c.relname IN ('validator_bigint', 'validator_uuid', 'config'))
     AND NOT c.relispartition
 ORDER BY
     n.nspname,

pgdog/src/backend/schema/postgres_fdw/schema.rs

@@ -1,7 +1,7 @@
 //! Foreign table schema query and data structures.
 
 use std::collections::{HashMap, HashSet};
-use tracing::debug;
+use tracing::{debug, warn};
 
 use crate::{
     backend::{schema::postgres_fdw::create_foreign_table, Server, ShardingSchema},
@@ -11,6 +11,7 @@ use crate::{
 use super::custom_types::CustomTypes;
 use super::extensions::Extensions;
 use super::quote_identifier;
+use super::TypeMismatch;
 
 /// Server definition for FDW setup.
 #[derive(Debug, Clone)]
@@ -134,18 +135,37 @@ impl ForeignTableSchema {
         self.custom_types.setup(server).await?;
 
         let mut tables = HashSet::new();
+        let mut all_type_mismatches: Vec<TypeMismatch> = Vec::new();
+
         for ((schema, table), columns) in &self.tables {
+            // Skip internal PgDog tables
+            if Self::is_internal_table(schema, table) {
+                continue;
+            }
+
             let dedup = (schema.clone(), table.clone());
             if !tables.contains(&dedup) {
-                let statements = create_foreign_table(columns, sharding_schema)?;
-                for sql in statements {
+                let result = create_foreign_table(columns, sharding_schema)?;
+                for sql in &result.statements {
                     debug!("[fdw::setup] {} [{}]", sql, server.addr());
-                    server.execute(&sql).await?;
+                    server.execute(sql).await?;
                 }
+                all_type_mismatches.extend(result.type_mismatches);
                 tables.insert(dedup);
             }
         }
 
+        // Log summary of type mismatches if any were found
+        if !all_type_mismatches.is_empty() {
+            warn!(
+                "[fdw] {} table(s) skipped due to sharding config type mismatches:",
+                all_type_mismatches.len()
+            );
+            for mismatch in &all_type_mismatches {
+                warn!("[fdw]   - {}", mismatch);
+            }
+        }
+
         server.execute("COMMIT").await?;
         Ok(())
     }
@@ -182,6 +202,11 @@ impl ForeignTableSchema {
         &self.custom_types
     }
 
+    /// Check if a table is an internal PgDog table that shouldn't be exposed via FDW.
+    fn is_internal_table(schema: &str, table: &str) -> bool {
+        schema == "pgdog" && matches!(table, "validator_bigint" | "validator_uuid" | "config")
+    }
+
     /// Collect unique schemas from tables and custom types.
     fn schemas(&self) -> HashSet<String> {
         self.tables