postgres_fdw fallback

Dockerfile

@@ -13,7 +13,7 @@ WORKDIR /build
 RUN rm /bin/sh && ln -s /bin/bash /bin/sh
 RUN source ~/.cargo/env && \
     if [ "$(uname -m)" = "aarch64" ] || [ "$(uname -m)" = "arm64" ]; then \
-        export RUSTFLAGS="-Ctarget-feature=+lse"; \
+    export RUSTFLAGS="-Ctarget-feature=+lse"; \
     fi && \
     cd pgdog && \
     cargo build --release
@@ -31,10 +31,13 @@ RUN install -d /usr/share/postgresql-common/pgdg && \
     . /etc/os-release && \
     sh -c "echo 'deb [signed-by=/usr/share/postgresql-common/pgdg/apt.postgresql.org.asc] https://apt.postgresql.org/pub/repos/apt $VERSION_CODENAME-pgdg main' > /etc/apt/sources.list.d/pgdg.list"
 
-RUN apt update && apt install -y postgresql-client-${PSQL_VERSION}
+RUN apt update && apt install -y postgresql-${PSQL_VERSION} && \
+    systemctl disable postgresql
 
 COPY --from=builder /build/target/release/pgdog /usr/local/bin/pgdog
 
+RUN mkdir -p /pgdog && chown postgres:postgres /pgdog
 WORKDIR /pgdog
+USER postgres
 STOPSIGNAL SIGINT
 CMD ["/usr/local/bin/pgdog"]

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 = "pgdog"
 
 [memory]
 net_buffer = 8096

integration/postgres_fdw/dev.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+set -ex -o pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PGDOG="$SCRIPT_DIR/../../target/debug/pgdog"
+
+dropdb shard_0_fdw || true
+dropdb shard_1_fdw || true
+
+createdb shard_0_fdw
+createdb shard_1_fdw
+
+psql -f "$SCRIPT_DIR/../schema_sync/ecommerce_schema.sql" shard_0_fdw
+psql -f "$SCRIPT_DIR/../schema_sync/ecommerce_schema.sql" shard_1_fdw
+
+${PGDOG}

integration/postgres_fdw/pgdog.toml

@@ -0,0 +1,36 @@
+
+[general]
+cross_shard_backend = "fdw"
+
+[[databases]]
+name = "pgdog"
+shard = 0
+host = "127.0.0.1"
+database_name = "shard_0_fdw"
+
+[[databases]]
+name = "pgdog"
+shard = 1
+host = "127.0.0.1"
+database_name = "shard_1_fdw"
+
+[[databases]]
+name = "pgdog"
+shard = 0
+host = "127.0.0.1"
+database_name = "shard_0_fdw"
+role = "replica"
+
+[[databases]]
+name = "pgdog"
+shard = 1
+host = "127.0.0.1"
+database_name = "shard_1_fdw"
+role = "replica"
+
+[[sharded_tables]]
+column = "user_id"
+database = "pgdog"
+
+[admin]
+password = "pgdog"

integration/postgres_fdw/users.toml

@@ -0,0 +1,9 @@
+[[users]]
+name = "pgdog"
+password = "pgdog"
+database = "pgdog"
+
+[[users]]
+name = "lev"
+password = "lev"
+database = "pgdog"

pgdog-config/src/core.rs

@@ -6,7 +6,7 @@ use tracing::{info, warn};
 
 use crate::sharding::ShardedSchema;
 use crate::{
-    system_catalogs, EnumeratedDatabase, Memory, OmnishardedTable, PassthoughAuth,
+    system_catalogs, EnumeratedDatabase, Fdw, Memory, OmnishardedTable, PassthoughAuth,
     PreparedStatements, QueryParserEngine, QueryParserLevel, ReadWriteSplit, RewriteMode, Role,
     SystemCatalogsBehavior,
 };
@@ -187,6 +187,9 @@ pub struct Config {
     /// Memory tweaks
     #[serde(default)]
     pub memory: Memory,
+
+    #[serde(default)]
+    pub fdw: Fdw,
 }
 
 impl Config {

pgdog-config/src/fdw.rs

@@ -0,0 +1,28 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Copy)]
+#[serde(deny_unknown_fields)]
+pub struct Fdw {
+    #[serde(default = "default_port")]
+    pub port: u16,
+
+    #[serde(default = "default_launch_timeout")]
+    pub launch_timeout: u64,
+}
+
+impl Default for Fdw {
+    fn default() -> Self {
+        Self {
+            port: default_port(),
+            launch_timeout: default_launch_timeout(),
+        }
+    }
+}
+
+fn default_port() -> u16 {
+    6433
+}
+
+fn default_launch_timeout() -> u64 {
+    5_000
+}

pgdog-config/src/general.rs

@@ -5,7 +5,9 @@ use std::path::PathBuf;
 use std::time::Duration;
 
 use crate::pooling::ConnectionRecovery;
-use crate::{CopyFormat, QueryParserEngine, QueryParserLevel, SystemCatalogsBehavior};
+use crate::{
+    CopyFormat, CrossShardBackend, QueryParserEngine, QueryParserLevel, SystemCatalogsBehavior,
+};
 
 use super::auth::{AuthType, PassthoughAuth};
 use super::database::{LoadBalancingStrategy, ReadWriteSplit, ReadWriteStrategy};
@@ -206,6 +208,9 @@ pub struct General {
     /// Trigger a schema reload on DDL like CREATE TABLE.
     #[serde(default = "General::reload_schema_on_ddl")]
     pub reload_schema_on_ddl: bool,
+    /// Cross-shard backend.
+    #[serde(default = "General::cross_shard_backend")]
+    pub cross_shard_backend: CrossShardBackend,
 }
 
 impl Default for General {
@@ -278,6 +283,7 @@ impl Default for General {
             omnisharded_sticky: bool::default(),
             resharding_copy_format: CopyFormat::default(),
             reload_schema_on_ddl: Self::reload_schema_on_ddl(),
+            cross_shard_backend: Self::cross_shard_backend(),
         }
     }
 }
@@ -406,6 +412,10 @@ impl General {
         )
     }
 
+    fn cross_shard_backend() -> CrossShardBackend {
+        Self::env_enum_or_default("PGDOG_CROSS_SHARD_BACKEND")
+    }
+
     pub fn query_timeout(&self) -> Duration {
         Duration::from_millis(self.query_timeout)
     }

pgdog-config/src/lib.rs

@@ -4,6 +4,7 @@ pub mod core;
 pub mod data_types;
 pub mod database;
 pub mod error;
+pub mod fdw;
 pub mod general;
 pub mod memory;
 pub mod networking;
@@ -24,6 +25,7 @@ pub use database::{
     Database, EnumeratedDatabase, LoadBalancingStrategy, ReadWriteSplit, ReadWriteStrategy, Role,
 };
 pub use error::Error;
+pub use fdw::Fdw;
 pub use general::General;
 pub use memory::*;
 pub use networking::{MultiTenant, Tcp, TlsVerifyMode};

pgdog-config/src/sharding.rs

@@ -313,6 +313,15 @@ impl ListShards {
             Ok(None)
         }
     }
+
+    /// Get all values that map to a specific shard.
+    pub fn values_for_shard(&self, shard: usize) -> Vec<&FlexibleType> {
+        self.mapping
+            .iter()
+            .filter(|(_, &s)| s == shard)
+            .map(|(v, _)| v)
+            .collect()
+    }
 }
 
 #[derive(Serialize, Deserialize, Debug, Copy, Clone, PartialEq, Eq, Hash, Default)]
@@ -370,6 +379,44 @@ impl Display for CopyFormat {
     }
 }
 
+#[derive(Serialize, Deserialize, Debug, Copy, Clone, PartialEq, Eq, Hash, Default)]
+#[serde(rename_all = "snake_case", deny_unknown_fields)]
+pub enum CrossShardBackend {
+    #[default]
+    Pgdog,
+    Fdw,
+    Hybrid,
+}
+
+impl CrossShardBackend {
+    pub fn need_fdw(&self) -> bool {
+        matches!(self, Self::Fdw | Self::Hybrid)
+    }
+}
+
+impl Display for CrossShardBackend {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Pgdog => write!(f, "pgdog"),
+            Self::Fdw => write!(f, "fdw"),
+            Self::Hybrid => write!(f, "hybrid"),
+        }
+    }
+}
+
+impl FromStr for CrossShardBackend {
+    type Err = ();
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "pgdog" => Ok(Self::Pgdog),
+            "fdw" => Ok(Self::Fdw),
+            "hybrid" => Ok(Self::Hybrid),
+            _ => Err(()),
+        }
+    }
+}
+
 #[cfg(test)]
 mod test {
     use super::*;

pgdog/Cargo.toml

@@ -62,6 +62,8 @@ hickory-resolver = "0.25.2"
 lazy_static = "1"
 dashmap = "6"
 derive_builder = "0.20.2"
+tempfile = "3.23.0"
+nix = { version = "0.31", features = ["signal"] }
 pgdog-config = { path = "../pgdog-config" }
 pgdog-vector = { path = "../pgdog-vector" }
 pgdog-stats = { path = "../pgdog-stats" }
@@ -75,5 +77,4 @@ tikv-jemallocator = "0.6"
 cc = "1"
 
 [dev-dependencies]
-tempfile = "3.23.0"
 stats_alloc = "0.1.10"

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),
         }