diff --git a/crates/etl-api/Cargo.toml b/crates/etl-api/Cargo.toml index 6403dde52..8ae9d2be2 100644 --- a/crates/etl-api/Cargo.toml +++ b/crates/etl-api/Cargo.toml @@ -18,12 +18,13 @@ path = "src/main.rs" name = "etl-api" [features] -default = ["bigquery", "clickhouse", "ducklake", "iceberg", "snowflake"] +default = ["bigquery", "clickhouse", "ducklake", "iceberg", "snowflake", "postgres"] bigquery = ["etl-destinations/bigquery"] clickhouse = ["etl-destinations/clickhouse"] ducklake = ["etl-destinations/ducklake"] iceberg = ["etl-destinations/iceberg"] snowflake = ["etl-destinations/snowflake"] +postgres = ["etl-destinations/postgres"] [dependencies] diff --git a/crates/etl-api/src/configs/destination.rs b/crates/etl-api/src/configs/destination.rs index a28a3cc04..f3016a472 100644 --- a/crates/etl-api/src/configs/destination.rs +++ b/crates/etl-api/src/configs/destination.rs @@ -1,6 +1,11 @@ +use std::net::IpAddr; + use etl_config::{ SerializableSecretString, - shared::{ClickHouseEngine, DestinationConfig, DuckLakeMaintenanceMode, IcebergConfig}, + shared::{ + ClickHouseEngine, DestinationConfig, DuckLakeMaintenanceMode, IcebergConfig, + PgConnectionConfig, TcpKeepaliveConfig, TlsConfig, + }, }; use secrecy::ExposeSecret; use serde::{Deserialize, Deserializer, Serialize, de::Error as _}; @@ -171,6 +176,31 @@ pub enum ApiDestinationConfig { )] role: Option, }, + Postgres { + #[schema(example = "localhost")] + #[serde(deserialize_with = "crate::utils::trim_string")] + host: String, + #[schema(value_type = String, example = "127.0.0.1")] + #[serde(skip_serializing_if = "Option::is_none")] + hostaddr: Option, + #[schema(example = 5432)] + port: u16, + #[schema(example = "analytics")] + #[serde(deserialize_with = "crate::utils::trim_string")] + name: String, + #[schema(example = "postgres")] + #[serde(deserialize_with = "crate::utils::trim_string")] + username: String, + #[serde(skip_serializing_if = "Option::is_none")] + password: Option, + #[schema(example = "replica")] + #[serde( + default, + skip_serializing_if = "Option::is_none", + deserialize_with = "crate::utils::trim_option_string" + )] + destination_schema: Option, + }, } /// Errors returned while merging destination update configuration. @@ -470,6 +500,49 @@ pub enum UpdateApiDestinationConfig { )] role: UpdateField, }, + + Postgres { + #[schema(example = "localhost")] + #[serde( + default, + skip_serializing_if = "UpdateField::is_preserve", + deserialize_with = "deserialize_update_trimmed_string" + )] + host: UpdateField, + #[serde(default, skip_serializing_if = "UpdateField::is_preserve")] + #[schema(value_type = Option, example = "127.0.0.1")] + hostaddr: UpdateField, + #[schema(example = 5432)] + #[serde(default, skip_serializing_if = "UpdateField::is_preserve")] + port: UpdateField, + #[schema(example = "analytics")] + #[serde( + default, + skip_serializing_if = "UpdateField::is_preserve", + deserialize_with = "deserialize_update_trimmed_string" + )] + name: UpdateField, + #[schema(example = "postgres")] + #[serde( + default, + skip_serializing_if = "UpdateField::is_preserve", + deserialize_with = "deserialize_update_trimmed_string" + )] + username: UpdateField, + #[serde( + default, + skip_serializing_if = "UpdateField::is_preserve", + deserialize_with = "deserialize_update_secret_string" + )] + password: UpdateField, + #[schema(example = "replica")] + #[serde( + default, + skip_serializing_if = "UpdateField::is_preserve", + deserialize_with = "deserialize_update_trimmed_string" + )] + destination_schema: UpdateField, + }, } impl UpdateApiDestinationConfig { @@ -552,6 +625,23 @@ impl UpdateApiDestinationConfig { schema: UpdateField::Set(schema), role: UpdateField::from_option(role), }, + ApiDestinationConfig::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + } => Self::Postgres { + host: UpdateField::Set(host), + hostaddr: UpdateField::from_option(hostaddr), + port: UpdateField::Set(port), + name: UpdateField::Set(name), + username: UpdateField::Set(username), + password: UpdateField::from_option(password), + destination_schema: UpdateField::from_option(destination_schema), + }, } } @@ -718,6 +808,41 @@ impl UpdateApiDestinationConfig { )?, role: role.apply_to_option(stored_role), }), + + ( + Self::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + }, + StoredDestinationConfig::Postgres { + host: stored_host, + hostaddr: stored_hostaddr, + port: stored_port, + name: stored_name, + username: stored_username, + password: stored_password, + destination_schema: stored_destination_schema, + }, + ) => Ok(StoredDestinationConfig::Postgres { + host: host + .apply_to_required(stored_host, required_field_cleared("Postgres", "host"))?, + hostaddr: hostaddr.apply_to_option(stored_hostaddr), + port: port + .apply_to_required(stored_port, required_field_cleared("Postgres", "port"))?, + name: name + .apply_to_required(stored_name, required_field_cleared("Postgres", "name"))?, + username: username.apply_to_required( + stored_username, + required_field_cleared("Postgres", "username"), + )?, + password: password.apply_to_option(stored_password), + destination_schema: destination_schema.apply_to_option(stored_destination_schema), + }), (config, _) => config.into_stored_requiring_secrets(), } } @@ -844,6 +969,36 @@ impl UpdateApiDestinationConfig { )?, role: role.apply_to_option(None), }), + + Self::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + } => Ok(StoredDestinationConfig::Postgres { + host: host.into_required( + missing_required_field("Postgres", "host"), + required_field_cleared("Postgres", "host"), + )?, + hostaddr: hostaddr.apply_to_option(None), + port: port.into_required( + missing_required_field("Postgres", "port"), + required_field_cleared("Postgres", "port"), + )?, + name: name.into_required( + missing_required_field("Postgres", "name"), + required_field_cleared("Postgres", "name"), + )?, + username: username.into_required( + missing_required_field("Postgres", "username"), + required_field_cleared("Postgres", "username"), + )?, + password: password.apply_to_option(None), + destination_schema: destination_schema.apply_to_option(None), + }), } } } @@ -975,6 +1130,24 @@ impl From for ApiDestinationConfig { schema, role, }, + + StoredDestinationConfig::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + } => Self::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + }, } } } @@ -1022,6 +1195,15 @@ pub enum StoredDestinationConfig { schema: String, role: Option, }, + Postgres { + host: String, + hostaddr: Option, + port: u16, + name: String, + username: String, + password: Option, + destination_schema: Option, + }, } impl StoredDestinationConfig { @@ -1133,6 +1315,30 @@ impl StoredDestinationConfig { schema, role, }, + + Self::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + } => DestinationConfig::Postgres { + pg_connection: PgConnectionConfig { + host, + hostaddr, + port, + name, + username, + password: password.map(Into::into), + // API-created Postgres destinations do not expose TLS fields yet. + // Library/replicator configs can enable TLS via `PgConnectionConfig.tls`. + tls: TlsConfig::disabled(), + keepalive: TcpKeepaliveConfig::default(), + }, + destination_schema, + }, } } } @@ -1241,6 +1447,24 @@ impl From for StoredDestinationConfig { schema, role, }, + + ApiDestinationConfig::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + } => Self::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + }, } } } @@ -1403,6 +1627,31 @@ impl Encrypt for StoredDestinationConfig { role, }) } + + Self::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + } => { + let password = password + .map(|password| { + encrypt_text(password.expose_secret().to_owned(), encryption_key) + }) + .transpose()?; + Ok(EncryptedStoredDestinationConfig::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + }) + } } } } @@ -1469,6 +1718,15 @@ pub enum EncryptedStoredDestinationConfig { #[serde(default, skip_serializing_if = "Option::is_none")] role: Option, }, + Postgres { + host: String, + hostaddr: Option, + port: u16, + name: String, + username: String, + password: Option, + destination_schema: Option, + }, } impl Store for EncryptedStoredDestinationConfig {} @@ -1646,6 +1904,30 @@ impl Decrypt for EncryptedStoredDestinationConfig { role, }) } + + Self::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + } => { + let password = password + .map(|password| decrypt_text(password, encryption_key)) + .transpose()? + .map(SerializableSecretString::from); + Ok(StoredDestinationConfig::Postgres { + host, + hostaddr, + port, + name, + username, + password, + destination_schema, + }) + } } } } diff --git a/crates/etl-api/src/k8s/base.rs b/crates/etl-api/src/k8s/base.rs index d66d9e7ab..7a412e7cd 100644 --- a/crates/etl-api/src/k8s/base.rs +++ b/crates/etl-api/src/k8s/base.rs @@ -100,6 +100,11 @@ pub enum DestinationType { /// secret entry. passphrase_secret_required: bool, }, + /// Postgres destination. + Postgres { + /// Whether the StatefulSet must reference the Postgres password secret. + password_secret_required: bool, + }, } impl DestinationType { @@ -112,6 +117,7 @@ impl DestinationType { DestinationType::ClickHouse { .. } => DestinationKind::ClickHouse, DestinationType::Ducklake => DestinationKind::Ducklake, DestinationType::Snowflake { .. } => DestinationKind::Snowflake, + DestinationType::Postgres { .. } => DestinationKind::Postgres, } } } @@ -131,6 +137,9 @@ impl From<&StoredDestinationConfig> for DestinationType { passphrase_secret_required: private_key_passphrase.is_some(), } } + StoredDestinationConfig::Postgres { password, .. } => { + DestinationType::Postgres { password_secret_required: password.is_some() } + } } } } diff --git a/crates/etl-api/src/k8s/core.rs b/crates/etl-api/src/k8s/core.rs index aeb6b957b..14ebb9794 100644 --- a/crates/etl-api/src/k8s/core.rs +++ b/crates/etl-api/src/k8s/core.rs @@ -95,6 +95,13 @@ pub enum Secrets { /// S3-compatible secret access key. s3_secret_access_key: String, }, + /// Credentials for Postgres destinations. + Postgres { + /// PostgreSQL source database password. + postgres_password: String, + /// Destination Postgres password. + password: Option, + }, /// Credentials for Snowflake destinations. Snowflake { /// PostgreSQL source database password. @@ -356,6 +363,10 @@ fn build_secrets_from_configs( .map(|p| p.expose_secret().to_owned()), } } + StoredDestinationConfig::Postgres { password, .. } => Secrets::Postgres { + postgres_password, + password: password.as_ref().map(|p| p.expose_secret().to_owned()), + }, }; Ok(secrets) @@ -435,6 +446,14 @@ async fn create_or_update_dynamic_replicator_secrets( k8s_client.delete_clickhouse_secret(prefix).await?; } } + Secrets::Postgres { postgres_password, password } => { + k8s_client.create_or_update_postgres_secret(prefix, &postgres_password).await?; + if let Some(password) = password.as_deref() { + k8s_client.create_or_update_clickhouse_secret(prefix, Some(password)).await?; + } else { + k8s_client.delete_clickhouse_secret(prefix).await?; + } + } Secrets::Ducklake { postgres_password, catalog_url, diff --git a/crates/etl-api/src/k8s/http.rs b/crates/etl-api/src/k8s/http.rs index 1eadccab5..140b96525 100644 --- a/crates/etl-api/src/k8s/http.rs +++ b/crates/etl-api/src/k8s/http.rs @@ -1448,6 +1448,19 @@ fn create_container_environment_json( .push(create_snowflake_passphrase_env_var_json(&snowflake_secret_name)); } } + DestinationType::Postgres { password_secret_required } => { + let postgres_secret_name = create_postgres_secret_name(prefix); + let postgres_secret_env_var_json = + create_postgres_secret_env_var_json(&postgres_secret_name); + container_environment.push(postgres_secret_env_var_json); + + if password_secret_required { + let destination_secret_name = create_clickhouse_secret_name(prefix); + container_environment.push(create_postgres_destination_password_env_var_json( + &destination_secret_name, + )); + } + } } container_environment } @@ -1625,6 +1638,18 @@ fn create_bq_secret_env_var_json(bq_secret_name: &str) -> serde_json::Value { }) } +fn create_postgres_destination_password_env_var_json(secret_name: &str) -> serde_json::Value { + json!({ + "name": "APP_DESTINATION__POSTGRES__PG_CONNECTION__PASSWORD", + "valueFrom": { + "secretKeyRef": { + "name": secret_name, + "key": CLICKHOUSE_PASSWORD_NAME + } + } + }) +} + fn create_clickhouse_secret_env_var_json(clickhouse_secret_name: &str) -> serde_json::Value { json!({ "name": "APP_DESTINATION__CLICKHOUSE__PASSWORD", diff --git a/crates/etl-api/src/validation/validators/destination.rs b/crates/etl-api/src/validation/validators/destination.rs index 849a00ec1..1c68b717a 100644 --- a/crates/etl-api/src/validation/validators/destination.rs +++ b/crates/etl-api/src/validation/validators/destination.rs @@ -61,6 +61,12 @@ impl DestinationValidator { &[IdentityType::Full], &[IdentityType::PrimaryKey, IdentityType::AlternativeKey, IdentityType::Full], ), + ApiDestinationConfig::Postgres { .. } => ReplicaIdentityValidator::new( + publication_name, + "Postgres", + &[IdentityType::PrimaryKey, IdentityType::Full], + &[IdentityType::PrimaryKey, IdentityType::Full], + ), }) } @@ -96,6 +102,12 @@ impl DestinationValidator { false, )) } + ApiDestinationConfig::Postgres { .. } => Some(PrimaryKeyValidator::new( + publication_name, + "Postgres", + "Postgres destination tables use the source primary key for UPSERT and DELETE application during initial loads and CDC.", + true, + )), _ => None, } } @@ -115,6 +127,7 @@ impl Validator for DestinationValidator { ApiDestinationConfig::Iceberg { .. } => iceberg::validate(&self.config, ctx).await, ApiDestinationConfig::Ducklake { .. } => ducklake::validate(&self.config, ctx).await, ApiDestinationConfig::Snowflake { .. } => snowflake::validate(&self.config, ctx).await, + ApiDestinationConfig::Postgres { .. } => postgres::validate(&self.config, ctx).await, }?; if let Some(validator) = self.replica_identity_validator() { @@ -325,3 +338,43 @@ mod snowflake { } disabled_destination!(snowflake, "snowflake", "Snowflake"); + +/// Postgres validation adapter. +#[cfg(feature = "postgres")] +mod postgres { + use super::{ApiDestinationConfig, ValidationContext, ValidationError, ValidationFailure}; + + /// Validates a Postgres destination configuration. + pub(super) async fn validate( + config: &ApiDestinationConfig, + _ctx: &ValidationContext, + ) -> Result, ValidationError> { + let ApiDestinationConfig::Postgres { host, name, username, .. } = config else { + unreachable!("Destination config should match Postgres."); + }; + + let mut failures = Vec::new(); + if host.trim().is_empty() { + failures.push(ValidationFailure::critical( + "Postgres Host Required", + "Enter the Postgres host that ETL should connect to.", + )); + } + if name.trim().is_empty() { + failures.push(ValidationFailure::critical( + "Postgres Database Required", + "Choose the Postgres database where replicated tables should be written.", + )); + } + if username.trim().is_empty() { + failures.push(ValidationFailure::critical( + "Postgres Username Required", + "Enter the Postgres user that ETL should connect with.", + )); + } + + Ok(failures) + } +} + +disabled_destination!(postgres, "postgres", "Postgres"); diff --git a/crates/etl-config/src/shared/destination.rs b/crates/etl-config/src/shared/destination.rs index 8fd4ff5d7..559952a2e 100644 --- a/crates/etl-config/src/shared/destination.rs +++ b/crates/etl-config/src/shared/destination.rs @@ -4,6 +4,8 @@ use url::Url; #[cfg(feature = "utoipa")] use utoipa::ToSchema; +use crate::shared::connection::{PgConnectionConfig, PgConnectionConfigWithoutSecrets}; + const fn default_connection_pool_size() -> usize { DestinationConfig::DEFAULT_CONNECTION_POOL_SIZE } @@ -76,6 +78,8 @@ pub enum DestinationKind { Iceberg, /// Snowflake destination. Snowflake, + /// Postgres destination. + Postgres, } impl DestinationKind { @@ -87,6 +91,7 @@ impl DestinationKind { DestinationKind::Ducklake => "ducklake", DestinationKind::Iceberg => "iceberg", DestinationKind::Snowflake => "snowflake", + DestinationKind::Postgres => "postgres", } } } @@ -195,6 +200,15 @@ pub enum DestinationConfig { /// Snowflake role. role: Option, }, + /// Postgres destination configuration. + Postgres { + /// Connection settings for the destination Postgres database. + pg_connection: PgConnectionConfig, + /// Optional schema override. When set, all tables are created in this + /// schema while preserving source table names. When omitted, source + /// `schema.table` names are preserved. + destination_schema: Option, + }, } impl DestinationConfig { @@ -211,6 +225,7 @@ impl DestinationConfig { DestinationConfig::Iceberg { .. } => DestinationKind::Iceberg, DestinationConfig::Ducklake { .. } => DestinationKind::Ducklake, DestinationConfig::Snowflake { .. } => DestinationKind::Snowflake, + DestinationConfig::Postgres { .. } => DestinationKind::Postgres, } } } @@ -413,6 +428,14 @@ pub enum DestinationConfigWithoutSecrets { #[serde(skip_serializing_if = "Option::is_none")] role: Option, }, + /// Postgres destination configuration without secrets. + Postgres { + /// Connection settings for the destination Postgres database. + pg_connection: PgConnectionConfigWithoutSecrets, + /// Optional schema override. + #[serde(skip_serializing_if = "Option::is_none")] + destination_schema: Option, + }, } impl From for DestinationConfigWithoutSecrets { @@ -477,6 +500,12 @@ impl From for DestinationConfigWithoutSecrets { schema, role, }, + DestinationConfig::Postgres { pg_connection, destination_schema } => { + DestinationConfigWithoutSecrets::Postgres { + pg_connection: pg_connection.into(), + destination_schema, + } + } } } } @@ -518,5 +547,6 @@ mod tests { assert_eq!(DestinationKind::Ducklake.as_str(), "ducklake"); assert_eq!(DestinationKind::Iceberg.as_str(), "iceberg"); assert_eq!(DestinationKind::Snowflake.as_str(), "snowflake"); + assert_eq!(DestinationKind::Postgres.as_str(), "postgres"); } } diff --git a/crates/etl-destinations/Cargo.toml b/crates/etl-destinations/Cargo.toml index 3d94e7c9e..7c6e3e4f6 100644 --- a/crates/etl-destinations/Cargo.toml +++ b/crates/etl-destinations/Cargo.toml @@ -83,9 +83,21 @@ snowflake = [ "dep:tracing", "dep:zstd", ] -# We assume that `test-utils` is always used in conjunction with `bigquery` or `iceberg` thus we only -# put here the extra dependencies needed. -test-utils = ["dep:uuid"] +postgres = [ + "dep:bytes", + "dep:etl-postgres", + "dep:parking_lot", + "dep:pg_escape", + "dep:rustls", + "dep:serde_json", + "dep:tokio", + "dep:tokio-postgres", + "dep:tracing", + "dep:uuid", +] +# We assume that `test-utils` is always used in conjunction with a destination +# feature; uuid is the shared extra dependency for test helpers. +test-utils = ["dep:uuid", "etl-postgres?/test-utils"] [dependencies] @@ -100,6 +112,7 @@ duckdb = { workspace = true, optional = true, features = ["appender-arrow", "bun etl = { workspace = true } etl-config = { workspace = true } etl-maintenance = { workspace = true, optional = true } +etl-postgres = { workspace = true, optional = true, features = ["tokio"] } futures = { workspace = true, optional = true } gcp-bigquery-client = { workspace = true, optional = true, features = ["rust-tls", "aws-lc-rs"] } humantime = { workspace = true, optional = true } @@ -116,6 +129,7 @@ r2d2 = { workspace = true, optional = true } rand = { workspace = true, optional = true, features = ["thread_rng"] } regex = { workspace = true, optional = true } reqwest = { workspace = true, optional = true, features = ["json", "rustls-tls"] } +rustls = { workspace = true, optional = true, features = ["aws-lc-rs", "logging"] } secrecy = { workspace = true, optional = true } serde = { workspace = true, optional = true, features = ["derive"] } serde_json = { workspace = true, optional = true, features = ["arbitrary_precision", "std"] } diff --git a/crates/etl-destinations/README.md b/crates/etl-destinations/README.md index 7ed3df4e4..4174144ce 100644 --- a/crates/etl-destinations/README.md +++ b/crates/etl-destinations/README.md @@ -11,6 +11,12 @@ Enable the destination modules you need with crate features: | `ducklake` | DuckLake | In progress | | `iceberg` | Apache Iceberg | Deprecated for now | | `snowflake` | Snowflake | In progress | +| `postgres` | Postgres | In progress | + +The Postgres destination creates current-state UPSERT tables (no CDC meta columns), +auto-applies portable schema changes, maps source `timetz` to destination `text`, +and accepts TLS through `PgConnectionConfig.tls` (API-created destinations currently +force TLS disabled). DuckLake external maintenance is configured at runtime with `maintenance_mode`: `disabled`, `kubernetes`, or `postgres`. The default is @@ -19,3 +25,5 @@ DuckLake external maintenance is configured at runtime with `ETL_DUCKLAKE_MAINTENANCE_CR_NAMESPACE`. Postgres coordination uses the same Postgres catalog connection as DuckLake and stores coordination state in the `etl` schema. + +When `destination_schema` is set, destination table names encode the source schema (for example `public.users` → `.public_users`) to avoid collisions across source schemas. diff --git a/crates/etl-destinations/src/lib.rs b/crates/etl-destinations/src/lib.rs index 8806a89ac..fd41522ea 100644 --- a/crates/etl-destinations/src/lib.rs +++ b/crates/etl-destinations/src/lib.rs @@ -15,6 +15,7 @@ mod sql; feature = "clickhouse", feature = "ducklake", feature = "iceberg", + feature = "postgres", feature = "snowflake" ))] mod table_name; @@ -27,5 +28,7 @@ pub mod clickhouse; pub mod ducklake; #[cfg(feature = "iceberg")] pub mod iceberg; +#[cfg(feature = "postgres")] +pub mod postgres; #[cfg(feature = "snowflake")] pub mod snowflake; diff --git a/crates/etl-destinations/src/postgres/client.rs b/crates/etl-destinations/src/postgres/client.rs new file mode 100644 index 000000000..9567aa0c7 --- /dev/null +++ b/crates/etl-destinations/src/postgres/client.rs @@ -0,0 +1,148 @@ +//! Postgres destination connection helpers. + +use std::sync::Arc; + +use etl::{ + error::{ErrorKind, EtlResult}, + etl_error, +}; +use etl_config::shared::{IntoConnectOptions, PgConnectionConfig}; +use etl_postgres::tokio::tls::MakeRustlsConnect; +use rustls::{ + ClientConfig, RootCertStore, + pki_types::{CertificateDer, pem::PemObject}, +}; +use tokio::sync::Mutex as TokioMutex; +use tokio_postgres::{Client, NoTls}; +use tracing::debug; + +use crate::postgres::encoding::map_postgres_error; + +/// Shared Postgres client with reconnect support. +#[derive(Clone)] +pub(crate) struct PostgresClient { + config: Arc, + inner: Arc>>, + /// Serializes reconnect attempts across clone-sharing workers. + reconnect_lock: Arc>, +} + +impl PostgresClient { + /// Creates a disconnected client handle. + pub(crate) fn new(config: PgConnectionConfig) -> Self { + Self { + config: Arc::new(config), + inner: Arc::new(TokioMutex::new(None)), + reconnect_lock: Arc::new(TokioMutex::new(())), + } + } + + /// Executes a DDL/DML statement without bind parameters. + pub(crate) async fn execute_simple( + &self, + sql: &str, + description: &'static str, + ) -> EtlResult<()> { + let mut guard = self.ensure_connected().await?; + let client = guard.as_mut().expect("connected client must be present"); + client.batch_execute(sql).await.map_err(|error| map_postgres_error(error, description))?; + Ok(()) + } + + /// Executes a parameterized statement. + pub(crate) async fn execute( + &self, + sql: &str, + params: &[&(dyn tokio_postgres::types::ToSql + Sync)], + description: &'static str, + ) -> EtlResult { + let mut guard = self.ensure_connected().await?; + let client = guard.as_mut().expect("connected client must be present"); + client.execute(sql, params).await.map_err(|error| map_postgres_error(error, description)) + } + + /// Ensures a live client is available, reconnecting when needed. + async fn ensure_connected(&self) -> EtlResult>> { + { + let guard = self.inner.lock().await; + if let Some(client) = guard.as_ref() + && !client.is_closed() + { + return Ok(guard); + } + } + + let _reconnect = self.reconnect_lock.lock().await; + let mut guard = self.inner.lock().await; + if let Some(client) = guard.as_ref() + && !client.is_closed() + { + return Ok(guard); + } + + debug!("connecting postgres destination client"); + let client = connect_postgres_client(&self.config).await?; + *guard = Some(client); + Ok(guard) + } +} + +/// Opens a tokio-postgres client using the shared TLS helper pattern. +pub(crate) async fn connect_postgres_client(config: &PgConnectionConfig) -> EtlResult { + let pg_config: tokio_postgres::Config = config.with_db(None); + + if config.tls.enabled { + let mut root_store = RootCertStore::empty(); + for cert in CertificateDer::pem_slice_iter(config.tls.trusted_root_certs.as_bytes()) { + let cert = cert.map_err(|error| { + etl_error!( + ErrorKind::DestinationConnectionFailed, + "Invalid Postgres destination TLS certificate", + source: error + ) + })?; + root_store.add(cert).map_err(|error| { + etl_error!( + ErrorKind::DestinationConnectionFailed, + "Failed to add Postgres destination TLS certificate", + source: error + ) + })?; + } + + let tls_config = + ClientConfig::builder().with_root_certificates(root_store).with_no_client_auth(); + let (client, connection) = + pg_config.connect(MakeRustlsConnect::new(tls_config)).await.map_err(|error| { + etl_error!( + ErrorKind::DestinationConnectionFailed, + "Failed to connect to Postgres destination", + source: error + ) + })?; + + tokio::spawn(async move { + if let Err(error) = connection.await { + tracing::info!(error = %error, "postgres destination connection closed"); + } + }); + + Ok(client) + } else { + let (client, connection) = pg_config.connect(NoTls).await.map_err(|error| { + etl_error!( + ErrorKind::DestinationConnectionFailed, + "Failed to connect to Postgres destination", + source: error + ) + })?; + + tokio::spawn(async move { + if let Err(error) = connection.await { + tracing::info!(error = %error, "postgres destination connection closed"); + } + }); + + Ok(client) + } +} diff --git a/crates/etl-destinations/src/postgres/core.rs b/crates/etl-destinations/src/postgres/core.rs new file mode 100644 index 000000000..45fb65f90 --- /dev/null +++ b/crates/etl-destinations/src/postgres/core.rs @@ -0,0 +1,877 @@ +//! Postgres destination implementation. + +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, +}; + +use etl::{ + bail, + data::{Cell, OldTableRow, TableRow, UpdatedTableRow}, + destination::{ + Destination, DestinationTableMetadata, DestinationTableSchemaStatus, + DestinationWriteStatus, DropTableForCopyResult, WriteEventsDurability, WriteEventsResult, + WriteTableRowsResult, + }, + error::{ErrorKind, EtlResult}, + etl_error, + event::Event, + schema::{ColumnSchema, IdentityType, ReplicatedTableSchema, SchemaDiff, TableId, TableName}, + store::{SchemaStore, StateStore}, +}; +use etl_config::shared::PgConnectionConfig; +use parking_lot::{Mutex, RwLock}; +use tracing::{debug, info, warn}; + +use crate::{ + postgres::{ + client::PostgresClient, + encoding::{cells_to_postgres_values, values_as_tosql_params}, + schema::{ + create_schema_sql, create_table_sql, delete_by_pk_sql, drop_table_sql, + ensure_has_primary_key, schema_diff_statements, truncate_table_sql, upsert_sql, + }, + }, + table_name::try_stringify_table_name, +}; + +/// Pending row operation for streaming CDC. +enum PendingOp { + Upsert(Vec), + Delete(Vec), +} + +/// CDC-capable Postgres destination that replicates tables with UPSERT +/// semantics. +#[derive(Clone)] +pub struct PostgresDestination { + client: PostgresClient, + store: Arc, + destination_schema: Option, + /// Table ids that have been ensured in this process. + ensured_tables: Arc>>, + /// Per-`table_id` locks serializing first-time table creation. + create_locks: Arc>>>>, +} + +impl PostgresDestination +where + S: StateStore + SchemaStore + Send + Sync, +{ + /// Creates a new Postgres destination. + /// + /// When `destination_schema` is `Some`, all tables are placed in that + /// schema and destination table names encode the source schema to avoid + /// collisions (`public.users` → `dest.public_users`). When `None`, source + /// `schema.table` names are preserved. + pub fn new( + pg_connection: PgConnectionConfig, + destination_schema: Option, + store: S, + ) -> Self { + Self { + client: PostgresClient::new(pg_connection), + store: Arc::new(store), + destination_schema, + ensured_tables: Arc::new(RwLock::new(HashSet::new())), + create_locks: Arc::new(Mutex::new(HashMap::new())), + } + } + + /// Resolves the destination [`TableName`] for a replicated source table. + /// When `destination_schema` is set, the source schema is encoded into the + /// destination table name (`schema_table` with underscore escaping) so tables + /// like `public.users` and `auth.users` cannot collide in one override schema. + fn destination_table_name(&self, schema: &ReplicatedTableSchema) -> EtlResult { + match &self.destination_schema { + Some(override_schema) => { + let encoded_table = try_stringify_table_name(schema.name())?; + Ok(TableName::new(override_schema.clone(), encoded_table)) + } + None => Ok(schema.name().clone()), + } + } + + /// Ensures the destination schema and table exist, recovering interrupted + /// DDL. + async fn ensure_table_exists(&self, schema: &ReplicatedTableSchema) -> EtlResult { + ensure_has_primary_key(schema)?; + let table_id = schema.id(); + let destination_table = self.destination_table_name(schema)?; + + if self.ensured_tables.read().contains(&table_id) { + return Ok(destination_table); + } + + let table_lock = { + let mut guard = self.create_locks.lock(); + Arc::clone(guard.entry(table_id).or_default()) + }; + let _create_guard = table_lock.lock().await; + + if self.ensured_tables.read().contains(&table_id) { + return Ok(destination_table); + } + + let destination_table_id = destination_table.as_quoted_identifier(); + + match self.store.get_destination_table_metadata(table_id).await? { + None => { + self.create_table_with_metadata( + table_id, + &destination_table, + &destination_table_id, + schema, + schema.inner().snapshot_id, + schema.replication_mask().clone(), + ) + .await?; + } + Some(metadata) => { + if metadata.is_applying() { + self.recover_applying_metadata(table_id, &destination_table, schema, metadata) + .await?; + } + } + } + + self.ensured_tables.write().insert(table_id); + Ok(destination_table) + } + + async fn create_table_with_metadata( + &self, + table_id: TableId, + destination_table: &TableName, + destination_table_id: &str, + schema: &ReplicatedTableSchema, + snapshot_id: etl::schema::SnapshotId, + replication_mask: etl::schema::ReplicationMask, + ) -> EtlResult<()> { + let metadata = DestinationTableMetadata::new_applying( + destination_table_id.to_owned(), + snapshot_id, + replication_mask, + ); + self.store.store_destination_table_metadata(table_id, metadata.clone()).await?; + self.issue_create_table(destination_table, schema).await?; + self.store.store_destination_table_metadata(table_id, metadata.to_applied()).await?; + Ok(()) + } + + async fn issue_create_table( + &self, + destination_table: &TableName, + schema: &ReplicatedTableSchema, + ) -> EtlResult<()> { + self.client + .execute_simple( + &create_schema_sql(&destination_table.schema), + "Postgres create schema failed", + ) + .await?; + let create_sql = create_table_sql(destination_table, schema)?; + self.client.execute_simple(&create_sql, "Postgres create table failed").await + } + + async fn recover_applying_metadata( + &self, + table_id: TableId, + destination_table: &TableName, + schema: &ReplicatedTableSchema, + metadata: DestinationTableMetadata, + ) -> EtlResult<()> { + warn!("table {} has Applying metadata, recovering interrupted operation", table_id); + + match metadata.previous_snapshot_id { + Some(prev_snapshot_id) => { + let old_table_schema = + self.store.get_table_schema(&table_id, prev_snapshot_id).await?.ok_or_else( + || { + etl_error!( + ErrorKind::InvalidState, + "Stored schema snapshot missing for Postgres schema recovery", + format!( + "Table {} needs stored schema snapshot {} to recover the \ + destination table, but it was not found.", + table_id, prev_snapshot_id + ) + ) + }, + )?; + let old_schema = ReplicatedTableSchema::from_mask( + old_table_schema, + metadata.replication_mask.clone(), + ); + let diff = old_schema.diff(schema); + self.apply_schema_diff(destination_table, &diff).await?; + } + None => { + self.issue_create_table(destination_table, schema).await?; + } + } + + self.store.store_destination_table_metadata(table_id, metadata.to_applied()).await?; + Ok(()) + } + + async fn handle_relation_event(&self, new_schema: &ReplicatedTableSchema) -> EtlResult<()> { + ensure_has_primary_key(new_schema)?; + + let table_id = new_schema.id(); + let new_snapshot_id = new_schema.inner().snapshot_id; + let new_replication_mask = new_schema.replication_mask().clone(); + + let metadata = + self.store.get_applied_destination_table_metadata(table_id).await?.ok_or_else( + || { + etl_error!( + ErrorKind::CorruptedTableSchema, + "Destination metadata missing for Postgres schema change", + format!( + "Table {} received schema snapshot {}, but destination metadata from \ + initial synchronization was not found.", + table_id, new_snapshot_id + ) + ) + }, + )?; + + let current_snapshot_id = metadata.snapshot_id; + let current_replication_mask = metadata.replication_mask.clone(); + + if current_snapshot_id == new_snapshot_id + && current_replication_mask == new_replication_mask + { + info!("schema for table {} unchanged (snapshot_id: {})", table_id, new_snapshot_id); + return Ok(()); + } + + info!( + "schema change detected for table {}: snapshot_id {} -> {}", + table_id, current_snapshot_id, new_snapshot_id + ); + + // Serialize with [`Self::ensure_table_exists`] so concurrent first-write + // CREATE and Relation DDL cannot race on the same destination table. + let table_lock = { + let mut guard = self.create_locks.lock(); + Arc::clone(guard.entry(table_id).or_default()) + }; + let _create_guard = table_lock.lock().await; + + let current_table_schema = + self.store.get_table_schema(&table_id, current_snapshot_id).await?.ok_or_else( + || { + etl_error!( + ErrorKind::InvalidState, + "Stored schema snapshot missing for Postgres schema change", + format!( + "Table {} needs stored schema snapshot {} to compare with incoming \ + snapshot {}, but it was not found.", + table_id, current_snapshot_id, new_snapshot_id + ) + ) + }, + )?; + + let current_schema = ReplicatedTableSchema::from_mask( + current_table_schema, + current_replication_mask.clone(), + ); + let destination_table = self.destination_table_name(new_schema)?; + let destination_table_id = destination_table.as_quoted_identifier(); + + let updated_metadata = DestinationTableMetadata::new_applied( + destination_table_id, + current_snapshot_id, + current_replication_mask, + ) + .with_schema_change( + new_snapshot_id, + new_replication_mask, + DestinationTableSchemaStatus::Applying, + ); + self.store.store_destination_table_metadata(table_id, updated_metadata.clone()).await?; + + let diff = current_schema.diff(new_schema); + self.apply_schema_diff(&destination_table, &diff).await?; + + self.store + .store_destination_table_metadata(table_id, updated_metadata.to_applied()) + .await?; + self.ensured_tables.write().insert(table_id); + + info!( + "schema change completed for table {}: snapshot_id {} applied", + table_id, new_snapshot_id + ); + + Ok(()) + } + + async fn apply_schema_diff( + &self, + destination_table: &TableName, + diff: &SchemaDiff, + ) -> EtlResult<()> { + for sql in schema_diff_statements(destination_table, diff) { + self.client.execute_simple(&sql, "Postgres alter table failed").await?; + } + Ok(()) + } + + async fn write_table_rows_inner( + &self, + schema: &ReplicatedTableSchema, + table_rows: Vec, + ) -> EtlResult<()> { + let destination_table = self.ensure_table_exists(schema).await?; + let sql = upsert_sql(&destination_table, schema)?; + + for table_row in table_rows { + let values = cells_to_postgres_values(table_row.into_values())?; + let params = values_as_tosql_params(&values); + self.client.execute(&sql, ¶ms, "Postgres upsert during table copy failed").await?; + } + + Ok(()) + } + + async fn upsert_row( + &self, + destination_table: &TableName, + schema: &ReplicatedTableSchema, + cells: Vec, + ) -> EtlResult<()> { + let sql = upsert_sql(destination_table, schema)?; + let values = cells_to_postgres_values(cells)?; + let params = values_as_tosql_params(&values); + self.client.execute(&sql, ¶ms, "Postgres upsert failed").await?; + Ok(()) + } + + async fn delete_row( + &self, + destination_table: &TableName, + schema: &ReplicatedTableSchema, + pk_cells: Vec, + ) -> EtlResult<()> { + let sql = delete_by_pk_sql(destination_table, schema)?; + let values = cells_to_postgres_values(pk_cells)?; + let params = values_as_tosql_params(&values); + self.client.execute(&sql, ¶ms, "Postgres delete failed").await?; + Ok(()) + } + + async fn truncate_table_inner(&self, schema: &ReplicatedTableSchema) -> EtlResult<()> { + let destination_table = self.ensure_table_exists(schema).await?; + self.client + .execute_simple(&truncate_table_sql(&destination_table), "Postgres truncate failed") + .await + } + + async fn drop_table_for_copy_inner(&self, schema: &ReplicatedTableSchema) -> EtlResult<()> { + let destination_table = self.destination_table_name(schema)?; + self.client + .execute_simple(&drop_table_sql(&destination_table), "Postgres drop table failed") + .await?; + self.ensured_tables.write().remove(&schema.id()); + Ok(()) + } + + async fn write_events_inner(&self, events: Vec) -> EtlResult<()> { + let mut event_iter = events.into_iter().peekable(); + + while event_iter.peek().is_some() { + let mut pending: HashMap)> = + HashMap::new(); + + while let Some(event) = event_iter.peek() { + if matches!(event, Event::Truncate(_) | Event::Relation(_)) { + break; + } + + let event = event_iter.next().expect("peeked event must be present"); + match event { + Event::Insert(insert) => { + let table_id = insert.replicated_table_schema.id(); + let entry = pending + .entry(table_id) + .or_insert_with(|| (insert.replicated_table_schema, Vec::new())); + entry.1.push(PendingOp::Upsert(insert.table_row.into_values())); + } + Event::Update(update) => { + let ops = postgres_update_ops( + &update.replicated_table_schema, + update.updated_table_row, + update.old_table_row, + )?; + let table_id = update.replicated_table_schema.id(); + let entry = pending + .entry(table_id) + .or_insert_with(|| (update.replicated_table_schema, Vec::new())); + entry.1.extend(ops); + } + Event::Delete(delete) => { + let pk_cells = postgres_delete_pk_cells( + &delete.replicated_table_schema, + delete.old_table_row, + )?; + let table_id = delete.replicated_table_schema.id(); + let entry = pending + .entry(table_id) + .or_insert_with(|| (delete.replicated_table_schema, Vec::new())); + entry.1.push(PendingOp::Delete(pk_cells)); + } + event => { + debug!(event_type = %event.event_type(), "skipping unsupported event type"); + } + } + } + + for (schema, ops) in pending.into_values() { + let destination_table = self.ensure_table_exists(&schema).await?; + for op in ops { + match op { + PendingOp::Upsert(cells) => { + self.upsert_row(&destination_table, &schema, cells).await?; + } + PendingOp::Delete(pk_cells) => { + self.delete_row(&destination_table, &schema, pk_cells).await?; + } + } + } + } + + while let Some(Event::Relation(_)) = event_iter.peek() { + if let Some(Event::Relation(relation)) = event_iter.next() { + self.handle_relation_event(&relation.replicated_table_schema).await?; + } + } + + let mut truncate_schemas: HashMap = HashMap::new(); + while let Some(Event::Truncate(_)) = event_iter.peek() { + if let Some(Event::Truncate(truncate_event)) = event_iter.next() { + for schema in truncate_event.truncated_tables { + truncate_schemas.entry(schema.id()).or_insert(schema); + } + } + } + + for schema in truncate_schemas.values() { + self.truncate_table_inner(schema).await?; + } + } + + Ok(()) + } +} + +impl Destination for PostgresDestination +where + S: StateStore + SchemaStore + Send + Sync, +{ + fn name() -> &'static str { + etl_config::shared::DestinationKind::Postgres.as_str() + } + + async fn drop_table_for_copy( + &self, + replicated_table_schema: &ReplicatedTableSchema, + async_result: DropTableForCopyResult<()>, + ) -> EtlResult<()> { + let result = self.drop_table_for_copy_inner(replicated_table_schema).await; + async_result.send(result); + Ok(()) + } + + async fn write_table_rows( + &self, + replicated_table_schema: &ReplicatedTableSchema, + table_rows: Vec, + async_result: WriteTableRowsResult, + ) -> EtlResult<()> { + let result = self.write_table_rows_inner(replicated_table_schema, table_rows).await; + async_result.send(result.map(|_| DestinationWriteStatus::Durable)); + Ok(()) + } + + async fn write_events( + &self, + events: Vec, + _durability: WriteEventsDurability, + async_result: WriteEventsResult, + ) -> EtlResult<()> { + let result = self.write_events_inner(events).await; + async_result.send(result.map(|_| DestinationWriteStatus::Durable)); + Ok(()) + } +} + +/// Builds pending delete/upsert ops for a Postgres update event. +fn postgres_update_ops( + replicated_table_schema: &ReplicatedTableSchema, + updated_table_row: UpdatedTableRow, + old_table_row: Option, +) -> EtlResult> { + let new_table_row = postgres_update_row(replicated_table_schema, updated_table_row)?; + let primary_key_changed = match old_table_row.as_ref() { + // PostgreSQL omits the old-side image only when the publisher + // determined it was unnecessary. For primary-key identity, that means + // the destination key did not change. `FULL` updates are expected to + // carry an old row from pgoutput. + Some(old_table_row) => { + postgres_primary_key_changed(replicated_table_schema, old_table_row, &new_table_row)? + } + None => { + ensure_postgres_update_without_old_row_can_skip_delete(replicated_table_schema)?; + false + } + }; + + let mut ops = Vec::with_capacity(1 + usize::from(primary_key_changed)); + if primary_key_changed { + let Some(old_table_row) = old_table_row else { + bail!( + ErrorKind::InvalidState, + "Postgres primary key change is missing old row", + format!( + "Table '{}' primary key change was detected without an old row image", + replicated_table_schema.name() + ) + ); + }; + + ops.push(PendingOp::Delete(postgres_delete_pk_cells( + replicated_table_schema, + Some(old_table_row), + )?)); + } + + ops.push(PendingOp::Upsert(new_table_row.into_values())); + Ok(ops) +} + +/// Returns the full new row required for a Postgres update upsert. +fn postgres_update_row( + replicated_table_schema: &ReplicatedTableSchema, + updated_table_row: UpdatedTableRow, +) -> EtlResult { + match updated_table_row { + UpdatedTableRow::Full(row) => Ok(row), + UpdatedTableRow::Partial(_) => Err(etl_error!( + ErrorKind::SourceReplicaIdentityError, + "Postgres update requires a full new row image", + format!( + "Table '{}' emitted a partial update row. Postgres UPSERT does not \ + preserve omitted columns.", + replicated_table_schema.name() + ) + )), + } +} + +/// Verifies that a Postgres update without an old row cannot have changed the +/// destination primary key. +fn ensure_postgres_update_without_old_row_can_skip_delete( + replicated_table_schema: &ReplicatedTableSchema, +) -> EtlResult<()> { + if matches!(replicated_table_schema.identity_type(), IdentityType::PrimaryKey) { + Ok(()) + } else { + Err(etl_error!( + ErrorKind::SourceReplicaIdentityError, + "Postgres update requires old primary-key values", + format!( + "Table '{}' emitted an update without an old row image for replica identity \ + {:?}. Postgres can only skip the generated delete when the \ + source replica identity matches the primary key.", + replicated_table_schema.name(), + replicated_table_schema.identity_type() + ) + )) + } +} + +/// Verifies that a key-only row image carries source primary-key values. +fn ensure_postgres_key_image_matches_primary_key( + replicated_table_schema: &ReplicatedTableSchema, +) -> EtlResult<()> { + if matches!(replicated_table_schema.identity_type(), IdentityType::PrimaryKey) { + Ok(()) + } else { + Err(etl_error!( + ErrorKind::SourceReplicaIdentityError, + "Postgres key image does not match the source primary key", + format!( + "Table '{}' emitted a key image for replica identity {:?}, but Postgres rows \ + are keyed by the source primary key", + replicated_table_schema.name(), + replicated_table_schema.identity_type() + ) + )) + } +} + +/// Returns whether an update changed the destination primary key. +fn postgres_primary_key_changed( + replicated_table_schema: &ReplicatedTableSchema, + old_table_row: &OldTableRow, + new_table_row: &TableRow, +) -> EtlResult { + let column_count = replicated_table_schema.column_schemas().len(); + if new_table_row.values().len() != column_count { + bail!( + ErrorKind::InvalidState, + "Postgres full row image does not match the replicated schema", + format!( + "Expected {} values for table '{}', got {}", + column_count, + replicated_table_schema.name(), + new_table_row.values().len() + ) + ); + } + + match old_table_row { + OldTableRow::Full(row) => { + if row.values().len() != column_count { + bail!( + ErrorKind::InvalidState, + "Postgres full row image does not match the replicated schema", + format!( + "Expected {} values for table '{}', got {}", + column_count, + replicated_table_schema.name(), + row.values().len() + ) + ); + } + + Ok(replicated_table_schema + .column_schemas() + .zip(row.values()) + .zip(new_table_row.values()) + .any(|((column_schema, old_value), new_value)| { + column_schema.primary_key() && old_value != new_value + })) + } + OldTableRow::Key(row) => { + let primary_key_column_count = + replicated_table_schema.primary_key_column_schemas().len(); + let old_key_values = row.values(); + if old_key_values.len() != primary_key_column_count { + bail!( + ErrorKind::InvalidState, + "Postgres key image does not match the source primary key", + format!( + "Expected {} key values for table '{}', got {}", + primary_key_column_count, + replicated_table_schema.name(), + old_key_values.len() + ) + ); + } + + ensure_postgres_key_image_matches_primary_key(replicated_table_schema)?; + + let mut new_primary_key_values = replicated_table_schema + .column_schemas() + .zip(new_table_row.values()) + .filter(|(column_schema, _)| column_schema.primary_key()) + .map(|(_, value)| value); + + for old_value in old_key_values { + let Some(new_value) = new_primary_key_values.next() else { + bail!( + ErrorKind::InvalidState, + "Postgres primary key schema mismatch", + format!( + "Table '{}' did not expose enough primary key columns", + replicated_table_schema.name() + ) + ); + }; + + if old_value != new_value { + return Ok(true); + } + } + + Ok(false) + } + } +} + +/// Extracts primary-key cells for a delete statement. +fn postgres_delete_pk_cells( + replicated_table_schema: &ReplicatedTableSchema, + old_table_row: Option, +) -> EtlResult> { + ensure_has_primary_key(replicated_table_schema)?; + + let old_table_row = old_table_row.ok_or_else(|| { + etl_error!( + ErrorKind::SourceReplicaIdentityError, + "Postgres delete requires an old row image", + format!( + "Table '{}' emitted a delete without an old row image.", + replicated_table_schema.name() + ) + ) + })?; + + match old_table_row { + OldTableRow::Full(row) => pk_cells_from_full_row(replicated_table_schema, row), + OldTableRow::Key(row) => pk_cells_from_key_row(replicated_table_schema, row), + } +} + +fn pk_cells_from_full_row(schema: &ReplicatedTableSchema, row: TableRow) -> EtlResult> { + let values = row.into_values(); + let pk_columns: Vec<&ColumnSchema> = schema.primary_key_column_schemas().collect(); + let replicated_columns: Vec<&ColumnSchema> = schema.column_schemas().collect(); + let mut pk_cells = Vec::with_capacity(pk_columns.len()); + for pk_column in pk_columns { + let index = replicated_columns + .iter() + .position(|column| column.ordinal_position == pk_column.ordinal_position) + .ok_or_else(|| { + etl_error!( + ErrorKind::InvalidState, + "Postgres delete could not locate primary-key column", + format!( + "Primary-key column '{}' missing from replicated schema for table '{}'", + pk_column.name, + schema.name() + ) + ) + })?; + pk_cells.push(values.get(index).cloned().unwrap_or(Cell::Null)); + } + + Ok(pk_cells) +} + +fn pk_cells_from_key_row(schema: &ReplicatedTableSchema, row: TableRow) -> EtlResult> { + match schema.identity_type() { + IdentityType::PrimaryKey | IdentityType::Full => {} + identity_type => { + return Err(etl_error!( + ErrorKind::SourceReplicaIdentityError, + "Postgres delete requires primary-key or full replica identity", + format!( + "Table '{}' uses replica identity {identity_type}. Configure REPLICA IDENTITY \ + DEFAULT or FULL so deletes can target the destination primary key.", + schema.name() + ) + )); + } + } + + let pk_columns: Vec<&ColumnSchema> = schema.primary_key_column_schemas().collect(); + let key_values = row.into_values(); + if key_values.len() != pk_columns.len() { + return Err(etl_error!( + ErrorKind::InvalidState, + "Postgres key image does not match the source primary key", + format!( + "Expected {} key values for table '{}', got {}", + pk_columns.len(), + schema.name(), + key_values.len() + ) + )); + } + + Ok(key_values) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use etl::{ + data::{Cell, PartialTableRow, TableRow, UpdatedTableRow}, + error::ErrorKind, + schema::{ + ColumnSchema, IdentityMask, IdentityType, ReplicatedTableSchema, TableId, TableName, + TableSchema, Type, + }, + }; + + use super::{PostgresDestination, postgres_update_row}; + + fn replicated_schema(identity_type: IdentityType) -> ReplicatedTableSchema { + let table_schema = Arc::new(TableSchema::new( + TableId::new(1), + TableName::new("public".to_owned(), "users".to_owned()), + vec![ + ColumnSchema::new("id".to_owned(), Type::INT4, -1, 1, false).with_primary_key(1), + ColumnSchema::new("name".to_owned(), Type::TEXT, -1, 2, true), + ], + )); + let replication_mask = etl::schema::ReplicationMask::all(&table_schema); + let identity_mask = match identity_type { + IdentityType::Full => IdentityMask::from_bytes(vec![1, 1]), + IdentityType::PrimaryKey => IdentityMask::from_bytes(vec![1, 0]), + IdentityType::AlternativeKey => IdentityMask::from_bytes(vec![0, 1]), + IdentityType::Missing => IdentityMask::from_bytes(vec![0, 0]), + }; + + ReplicatedTableSchema::from_masks(table_schema, replication_mask, identity_mask) + } + + #[test] + fn postgres_update_row_rejects_partial_rows() { + let replicated_table_schema = replicated_schema(IdentityType::PrimaryKey); + let partial_row = PartialTableRow::new(2, TableRow::new(vec![Cell::I32(1)]), vec![1]); + + let error = + postgres_update_row(&replicated_table_schema, UpdatedTableRow::Partial(partial_row)) + .unwrap_err(); + + assert_eq!(error.kind(), ErrorKind::SourceReplicaIdentityError); + assert!(error.to_string().contains("emitted a partial update row")); + } + #[test] + fn destination_table_name_encodes_source_schema_under_override() { + use etl::store::MemoryStore; + + let store = MemoryStore::new(); + let destination = PostgresDestination::new( + etl_config::shared::PgConnectionConfig { + host: "localhost".into(), + hostaddr: None, + port: 5432, + name: "postgres".into(), + username: "postgres".into(), + password: None, + tls: etl_config::shared::TlsConfig::disabled(), + keepalive: etl_config::shared::TcpKeepaliveConfig::default(), + }, + Some("replica".to_owned()), + store, + ); + + let public_users = ReplicatedTableSchema::all(Arc::new(TableSchema::new( + TableId::new(1), + TableName::new("public".to_owned(), "users".to_owned()), + vec![ColumnSchema::new("id".to_owned(), Type::INT4, -1, 1, false).with_primary_key(1)], + ))); + let auth_users = ReplicatedTableSchema::all(Arc::new(TableSchema::new( + TableId::new(2), + TableName::new("auth".to_owned(), "users".to_owned()), + vec![ColumnSchema::new("id".to_owned(), Type::INT4, -1, 1, false).with_primary_key(1)], + ))); + + let public_dest = destination.destination_table_name(&public_users).unwrap(); + let auth_dest = destination.destination_table_name(&auth_users).unwrap(); + assert_eq!(public_dest.schema, "replica"); + assert_eq!(auth_dest.schema, "replica"); + assert_eq!(public_dest.name, "public_users"); + assert_eq!(auth_dest.name, "auth_users"); + assert_ne!(public_dest.name, auth_dest.name); + } +} diff --git a/crates/etl-destinations/src/postgres/encoding.rs b/crates/etl-destinations/src/postgres/encoding.rs new file mode 100644 index 000000000..31aafe795 --- /dev/null +++ b/crates/etl-destinations/src/postgres/encoding.rs @@ -0,0 +1,178 @@ +//! Cell encoding for Postgres destination writes. + +use std::error::Error; + +use bytes::BytesMut; +use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc}; +use etl::{ + data::{ArrayCell, Cell, PgNumeric}, + error::{ErrorKind, EtlResult}, + etl_error, +}; +use tokio_postgres::types::{IsNull, ToSql, Type, to_sql_checked}; +use uuid::Uuid; + +/// A Postgres-bound value converted from an ETL [`Cell`]. +#[derive(Debug, Clone)] +pub(crate) enum PostgresValue { + Null, + Bool(bool), + String(String), + I16(i16), + I32(i32), + U32(u32), + I64(i64), + F32(f32), + F64(f64), + Numeric(PgNumeric), + Date(NaiveDate), + Time(NaiveTime), + /// Source `timetz` rendered as text; destination DDL maps `timetz` to + /// `text`. + TimeTz(String), + Timestamp(NaiveDateTime), + TimestampTz(DateTime), + Uuid(Uuid), + Json(serde_json::Value), + Bytes(Vec), + BoolArray(Vec>), + StringArray(Vec>), + I16Array(Vec>), + I32Array(Vec>), + U32Array(Vec>), + I64Array(Vec>), + F32Array(Vec>), + F64Array(Vec>), + NumericArray(Vec>), + DateArray(Vec>), + TimeArray(Vec>), + TimeTzArray(Vec>), + TimestampArray(Vec>), + TimestampTzArray(Vec>>), + UuidArray(Vec>), + JsonArray(Vec>), + BytesArray(Vec>>), +} + +impl PostgresValue { + /// Converts an ETL cell into a bindable Postgres value. + pub(crate) fn from_cell(cell: Cell) -> EtlResult { + Ok(match cell { + Cell::Null => Self::Null, + Cell::Bool(value) => Self::Bool(value), + Cell::String(value) => Self::String(value), + Cell::I16(value) => Self::I16(value), + Cell::I32(value) => Self::I32(value), + Cell::U32(value) => Self::U32(value), + Cell::I64(value) => Self::I64(value), + Cell::F32(value) => Self::F32(value), + Cell::F64(value) => Self::F64(value), + Cell::Numeric(value) => Self::Numeric(value), + Cell::Date(value) => Self::Date(value), + Cell::Time(value) => Self::Time(value), + Cell::TimeTz(value) => Self::TimeTz(value.to_string()), + Cell::Timestamp(value) => Self::Timestamp(value), + Cell::TimestampTz(value) => Self::TimestampTz(value), + Cell::Uuid(value) => Self::Uuid(value), + Cell::Json(value) => Self::Json(value), + Cell::Bytes(value) => Self::Bytes(value), + Cell::Array(array) => Self::from_array_cell(array)?, + }) + } + + fn from_array_cell(array: ArrayCell) -> EtlResult { + Ok(match array { + ArrayCell::Bool(values) => Self::BoolArray(values), + ArrayCell::String(values) => Self::StringArray(values), + ArrayCell::I16(values) => Self::I16Array(values), + ArrayCell::I32(values) => Self::I32Array(values), + ArrayCell::U32(values) => Self::U32Array(values), + ArrayCell::I64(values) => Self::I64Array(values), + ArrayCell::F32(values) => Self::F32Array(values), + ArrayCell::F64(values) => Self::F64Array(values), + ArrayCell::Numeric(values) => Self::NumericArray(values), + ArrayCell::Date(values) => Self::DateArray(values), + ArrayCell::Time(values) => Self::TimeArray(values), + ArrayCell::TimeTz(values) => Self::TimeTzArray( + values.into_iter().map(|value| value.map(|inner| inner.to_string())).collect(), + ), + ArrayCell::Timestamp(values) => Self::TimestampArray(values), + ArrayCell::TimestampTz(values) => Self::TimestampTzArray(values), + ArrayCell::Uuid(values) => Self::UuidArray(values), + ArrayCell::Json(values) => Self::JsonArray(values), + ArrayCell::Bytes(values) => Self::BytesArray(values), + }) + } +} + +impl ToSql for PostgresValue { + fn to_sql( + &self, + ty: &Type, + out: &mut BytesMut, + ) -> Result> { + match self { + Self::Null => Ok(IsNull::Yes), + Self::Bool(value) => value.to_sql(ty, out), + Self::String(value) => value.to_sql(ty, out), + Self::I16(value) => value.to_sql(ty, out), + Self::I32(value) => value.to_sql(ty, out), + Self::U32(value) => value.to_sql(ty, out), + Self::I64(value) => value.to_sql(ty, out), + Self::F32(value) => value.to_sql(ty, out), + Self::F64(value) => value.to_sql(ty, out), + Self::Numeric(value) => value.to_sql(ty, out), + Self::Date(value) => value.to_sql(ty, out), + Self::Time(value) => value.to_sql(ty, out), + Self::TimeTz(value) => value.to_sql(ty, out), + Self::Timestamp(value) => value.to_sql(ty, out), + Self::TimestampTz(value) => value.to_sql(ty, out), + Self::Uuid(value) => value.to_sql(ty, out), + Self::Json(value) => value.to_sql(ty, out), + Self::Bytes(value) => value.to_sql(ty, out), + Self::BoolArray(values) => values.to_sql(ty, out), + Self::StringArray(values) => values.to_sql(ty, out), + Self::I16Array(values) => values.to_sql(ty, out), + Self::I32Array(values) => values.to_sql(ty, out), + Self::U32Array(values) => values.to_sql(ty, out), + Self::I64Array(values) => values.to_sql(ty, out), + Self::F32Array(values) => values.to_sql(ty, out), + Self::F64Array(values) => values.to_sql(ty, out), + Self::NumericArray(values) => values.to_sql(ty, out), + Self::DateArray(values) => values.to_sql(ty, out), + Self::TimeArray(values) => values.to_sql(ty, out), + Self::TimeTzArray(values) => values.to_sql(ty, out), + Self::TimestampArray(values) => values.to_sql(ty, out), + Self::TimestampTzArray(values) => values.to_sql(ty, out), + Self::UuidArray(values) => values.to_sql(ty, out), + Self::JsonArray(values) => values.to_sql(ty, out), + Self::BytesArray(values) => values.to_sql(ty, out), + } + } + + fn accepts(ty: &Type) -> bool { + // Destination DDL maps source `timetz` / `timetz[]` columns to `text` / + // `text[]`, so reject native timetz bind targets. + !matches!(*ty, Type::TIMETZ | Type::TIMETZ_ARRAY) + } + + to_sql_checked!(); +} + +/// Converts a row of cells into Postgres bind values. +pub(crate) fn cells_to_postgres_values(cells: Vec) -> EtlResult> { + cells.into_iter().map(PostgresValue::from_cell).collect() +} + +/// Converts cells into trait-object references for `tokio-postgres` binds. +pub(crate) fn values_as_tosql_params(values: &[PostgresValue]) -> Vec<&(dyn ToSql + Sync)> { + values.iter().map(|value| value as &(dyn ToSql + Sync)).collect() +} + +/// Maps a destination client/query failure into an [`etl::error::EtlError`]. +pub(crate) fn map_postgres_error( + error: tokio_postgres::Error, + description: &'static str, +) -> etl::error::EtlError { + etl_error!(ErrorKind::DestinationQueryFailed, description, source: error) +} diff --git a/crates/etl-destinations/src/postgres/mod.rs b/crates/etl-destinations/src/postgres/mod.rs new file mode 100644 index 000000000..a0c3e24e6 --- /dev/null +++ b/crates/etl-destinations/src/postgres/mod.rs @@ -0,0 +1,16 @@ +//! Postgres destination with automatic schema management. +//! +//! Replicates Postgres tables into another Postgres database using +//! current-state UPSERT tables (no CDC meta columns). Destination schemas and +//! tables are created on first write, and supported schema changes are applied +//! from [`etl::event::Event::Relation`] events. + +mod client; +mod core; +mod encoding; +mod schema; +mod sql; +#[cfg(feature = "test-utils")] +pub mod test_utils; + +pub use core::PostgresDestination; diff --git a/crates/etl-destinations/src/postgres/schema.rs b/crates/etl-destinations/src/postgres/schema.rs new file mode 100644 index 000000000..e0cce3c21 --- /dev/null +++ b/crates/etl-destinations/src/postgres/schema.rs @@ -0,0 +1,473 @@ +//! DDL generation for the Postgres destination. + +use std::borrow::Cow; + +use etl::{ + error::{ErrorKind, EtlResult}, + etl_error, + schema::{ + ColumnModification, ColumnSchema, DefaultExpression, NumericModifiers, + ReplicatedTableSchema, SchemaDiff, Type, is_array_type, numeric_modifiers, + parse_default_expression, + }, +}; +use tracing::warn; + +use crate::postgres::sql::{quote_identifier, quote_table_name}; + +/// Postgres VARHDRSZ used in typmod encoding for variable-length types. +const VARHDRSZ: i32 = 4; + +/// Builds `CREATE SCHEMA IF NOT EXISTS` for a schema name. +pub(crate) fn create_schema_sql(schema_name: &str) -> String { + format!("create schema if not exists {}", quote_identifier(schema_name)) +} + +/// Builds `CREATE TABLE IF NOT EXISTS` for a replicated schema. +pub(crate) fn create_table_sql( + destination_table: &etl::schema::TableName, + schema: &ReplicatedTableSchema, +) -> EtlResult { + ensure_has_primary_key(schema)?; + + let mut column_defs = Vec::new(); + for column in schema.column_schemas() { + column_defs.push(column_definition(column, false, true, true)); + } + + let mut pk_columns: Vec<_> = schema.primary_key_column_schemas().collect(); + pk_columns.sort_by_key(|column| column.primary_key_ordinal_position); + let pk_list = pk_columns + .iter() + .map(|column| quote_identifier(&column.name)) + .collect::>() + .join(", "); + + Ok(format!( + "create table if not exists {} ({}, primary key ({}))", + quote_table_name(destination_table), + column_defs.join(", "), + pk_list + )) +} + +/// Builds DDL statements that apply a [`SchemaDiff`] to an existing table. +/// +/// Statement order is intentional for name-reuse collisions such as renaming +/// column `a` to `b` while also adding a new column named `a`: +/// 1. DROP removed columns +/// 2. RENAME columns +/// 3. ADD new columns +/// 4. nullability / default modifications +pub(crate) fn schema_diff_statements( + destination_table: &etl::schema::TableName, + diff: &SchemaDiff, +) -> Vec { + if diff.is_empty() { + return Vec::new(); + } + + let table = quote_table_name(destination_table); + let mut statements = Vec::new(); + + for column in &diff.columns_to_remove { + statements.push(format!( + "alter table {table} drop column if exists {}", + quote_identifier(&column.name) + )); + } + + for change in &diff.columns_to_change { + for modification in &change.modifications { + let ColumnModification::Rename { old_name, new_name } = modification else { + continue; + }; + statements.push(format!( + "alter table {table} rename column {} to {}", + quote_identifier(old_name), + quote_identifier(new_name) + )); + } + } + + for column in &diff.columns_to_add { + // Existing rows cannot satisfy a new NOT NULL without a default, so force + // nullability unless a supported default expression is present. + let force_nullable = column.default_expression.is_none() + || !supports_column_default( + column.default_expression.as_deref().unwrap_or_default(), + &column.typ, + ); + statements.push(format!( + "alter table {table} add column {}", + column_definition(column, force_nullable, true, true) + )); + } + + for change in &diff.columns_to_change { + for modification in &change.modifications { + match modification { + ColumnModification::Rename { .. } => {} + ColumnModification::Nullability { old_nullable, new_nullable } => { + if !*old_nullable && *new_nullable { + statements.push(format!( + "alter table {table} alter column {} drop not null", + quote_identifier(&change.new_column.name) + )); + } else if *old_nullable && !*new_nullable { + warn!( + table_name = %table, + column_name = %change.new_column.name, + "skipping source column set not null for Postgres destination" + ); + } + } + ColumnModification::Default { old_expression, new_expression } => { + let old_default_was_supported = + old_expression.as_deref().is_some_and(|expression| { + supports_column_default(expression, &change.old_column.typ) + }); + + if let Some(new_default_expression) = new_expression.as_deref() { + if let Some(rendered) = postgres_default_expression( + new_default_expression, + &change.new_column.typ, + ) { + statements.push(format!( + "alter table {table} alter column {} set default {rendered}", + quote_identifier(&change.new_column.name) + )); + } else { + warn!( + table_name = %table, + column_name = %change.new_column.name, + "skipping unsupported source column default for Postgres destination" + ); + if old_default_was_supported { + statements.push(format!( + "alter table {table} alter column {} drop default", + quote_identifier(&change.new_column.name) + )); + } + } + } else if old_default_was_supported { + statements.push(format!( + "alter table {table} alter column {} drop default", + quote_identifier(&change.new_column.name) + )); + } else if old_expression.is_some() { + warn!( + table_name = %table, + column_name = %change.new_column.name, + "skipping source column default removal for Postgres destination because no supported destination default was set" + ); + } + } + } + } + } + + statements +} + +/// Builds `DROP TABLE IF EXISTS` for a destination table. +pub(crate) fn drop_table_sql(destination_table: &etl::schema::TableName) -> String { + format!("drop table if exists {}", quote_table_name(destination_table)) +} + +/// Builds `TRUNCATE TABLE` for a destination table. +pub(crate) fn truncate_table_sql(destination_table: &etl::schema::TableName) -> String { + format!("truncate table {}", quote_table_name(destination_table)) +} + +/// Builds an upsert statement for the given schema columns. +pub(crate) fn upsert_sql( + destination_table: &etl::schema::TableName, + schema: &ReplicatedTableSchema, +) -> EtlResult { + ensure_has_primary_key(schema)?; + + let columns: Vec<_> = schema.column_schemas().collect(); + let column_list = + columns.iter().map(|column| quote_identifier(&column.name)).collect::>().join(", "); + let placeholders = + (1..=columns.len()).map(|index| format!("${index}")).collect::>().join(", "); + + let mut pk_columns: Vec<_> = schema.primary_key_column_schemas().collect(); + pk_columns.sort_by_key(|column| column.primary_key_ordinal_position); + let pk_list = pk_columns + .iter() + .map(|column| quote_identifier(&column.name)) + .collect::>() + .join(", "); + + let update_assignments = columns + .iter() + .filter(|column| column.primary_key_ordinal_position.is_none()) + .map(|column| { + let name = quote_identifier(&column.name); + format!("{name} = excluded.{name}") + }) + .collect::>(); + + if update_assignments.is_empty() { + Ok(format!( + "insert into {} ({column_list}) values ({placeholders}) on conflict ({pk_list}) do \ + nothing", + quote_table_name(destination_table) + )) + } else { + Ok(format!( + "insert into {} ({column_list}) values ({placeholders}) on conflict ({pk_list}) do \ + update set {}", + quote_table_name(destination_table), + update_assignments.join(", ") + )) + } +} + +/// Builds a delete-by-primary-key statement. +pub(crate) fn delete_by_pk_sql( + destination_table: &etl::schema::TableName, + schema: &ReplicatedTableSchema, +) -> EtlResult { + ensure_has_primary_key(schema)?; + + let pk_columns: Vec<_> = schema.primary_key_column_schemas().collect(); + let predicates = pk_columns + .iter() + .enumerate() + .map(|(index, column)| format!("{} = ${}", quote_identifier(&column.name), index + 1)) + .collect::>() + .join(" and "); + + Ok(format!("delete from {} where {predicates}", quote_table_name(destination_table))) +} + +/// Returns whether a table has at least one replicated primary-key column. +pub(crate) fn ensure_has_primary_key(schema: &ReplicatedTableSchema) -> EtlResult<()> { + if schema.primary_key_column_schemas().len() == 0 { + return Err(etl_error!( + ErrorKind::ValidationError, + "Postgres destination requires a primary key", + format!( + "Table '{}' has no replicated primary-key columns. The Postgres destination uses \ + UPSERT tables keyed by the source primary key.", + schema.name() + ) + )); + } + + if !schema.all_primary_key_columns_replicated() { + let missing = schema + .unreplicated_primary_key_column_schemas() + .map(|column| column.name.as_str()) + .collect::>() + .join(", "); + return Err(etl_error!( + ErrorKind::ValidationError, + "Postgres destination requires all primary-key columns to be replicated", + format!( + "Table '{}' omits primary-key columns from replication: {missing}", + schema.name() + ) + )); + } + + Ok(()) +} + +/// Builds one column definition for CREATE/ALTER TABLE. +fn column_definition( + column: &ColumnSchema, + force_nullable: bool, + include_default: bool, + include_not_null: bool, +) -> String { + let sql_type = postgres_column_type_sql(&column.typ, column.modifier); + let default_clause = if include_default { + column + .default_expression + .as_deref() + .and_then(|expression| postgres_default_expression(expression, &column.typ)) + .map(|rendered| format!(" default {rendered}")) + .unwrap_or_default() + } else { + String::new() + }; + let nullable = force_nullable || column.nullable || !include_not_null; + let nullability = if nullable { "" } else { " not null" }; + format!("{} {sql_type}{default_clause}{nullability}", quote_identifier(&column.name)) +} + +/// Returns the Postgres SQL type for a column type and modifier. +pub(crate) fn postgres_column_type_sql(typ: &Type, modifier: i32) -> Cow<'static, str> { + if is_array_type(typ) { + let element = array_element_type_sql(typ, modifier); + format!("{element}[]").into() + } else { + scalar_type_sql(typ, modifier) + } +} + +fn scalar_type_sql(typ: &Type, modifier: i32) -> Cow<'static, str> { + match *typ { + Type::BOOL => "boolean".into(), + Type::INT2 => "smallint".into(), + Type::INT4 => "integer".into(), + Type::INT8 => "bigint".into(), + Type::FLOAT4 => "real".into(), + Type::FLOAT8 => "double precision".into(), + Type::NUMERIC => match numeric_modifiers(modifier) { + Some(NumericModifiers { p, s }) if s >= 0 => format!("numeric({p}, {s})").into(), + Some(NumericModifiers { p, .. }) => format!("numeric({p})").into(), + None => "numeric".into(), + }, + Type::TEXT | Type::NAME => "text".into(), + Type::VARCHAR => match char_length(modifier) { + Some(length) => format!("varchar({length})").into(), + None => "varchar".into(), + }, + Type::BPCHAR | Type::CHAR => match char_length(modifier) { + Some(length) => format!("character({length})").into(), + None => "character".into(), + }, + Type::DATE => "date".into(), + Type::TIME => "time".into(), + // Bound as text via PostgresValue; keep the destination column as text so + // prepared-statement parameter types match the encoded values. + Type::TIMETZ => "text".into(), + Type::TIMESTAMP => "timestamp".into(), + Type::TIMESTAMPTZ => "timestamptz".into(), + Type::UUID => "uuid".into(), + Type::JSON => "json".into(), + Type::JSONB => "jsonb".into(), + Type::BYTEA => "bytea".into(), + Type::OID => "oid".into(), + Type::INET => "inet".into(), + Type::CIDR => "cidr".into(), + Type::MACADDR => "macaddr".into(), + Type::INTERVAL => "interval".into(), + _ => "text".into(), + } +} + +fn array_element_type_sql(typ: &Type, modifier: i32) -> Cow<'static, str> { + match *typ { + Type::BOOL_ARRAY => "boolean".into(), + Type::INT2_ARRAY => "smallint".into(), + Type::INT4_ARRAY => "integer".into(), + Type::INT8_ARRAY => "bigint".into(), + Type::FLOAT4_ARRAY => "real".into(), + Type::FLOAT8_ARRAY => "double precision".into(), + Type::NUMERIC_ARRAY => match numeric_modifiers(modifier) { + Some(NumericModifiers { p, s }) if s >= 0 => format!("numeric({p}, {s})").into(), + Some(NumericModifiers { p, .. }) => format!("numeric({p})").into(), + None => "numeric".into(), + }, + Type::TEXT_ARRAY => "text".into(), + Type::VARCHAR_ARRAY => match char_length(modifier) { + Some(length) => format!("varchar({length})").into(), + None => "varchar".into(), + }, + Type::BPCHAR_ARRAY => match char_length(modifier) { + Some(length) => format!("character({length})").into(), + None => "character".into(), + }, + Type::DATE_ARRAY => "date".into(), + Type::TIME_ARRAY => "time".into(), + Type::TIMETZ_ARRAY => "text".into(), + Type::TIMESTAMP_ARRAY => "timestamp".into(), + Type::TIMESTAMPTZ_ARRAY => "timestamptz".into(), + Type::UUID_ARRAY => "uuid".into(), + Type::JSON_ARRAY => "json".into(), + Type::JSONB_ARRAY => "jsonb".into(), + Type::BYTEA_ARRAY => "bytea".into(), + Type::OID_ARRAY => "oid".into(), + Type::INET_ARRAY => "inet".into(), + Type::CIDR_ARRAY => "cidr".into(), + Type::MACADDR_ARRAY => "macaddr".into(), + Type::INTERVAL_ARRAY => "interval".into(), + _ => "text".into(), + } +} + +/// Returns VARCHAR/BPCHAR character length from typmod, if constrained. +fn char_length(modifier: i32) -> Option { + if modifier == -1 { None } else { Some(modifier - VARHDRSZ) } +} + +/// Returns whether a default expression can be applied on Postgres. +pub(crate) fn supports_column_default(default_expression: &str, typ: &Type) -> bool { + postgres_default_expression(default_expression, typ).is_some() +} + +/// Renders a supported default expression as Postgres SQL. +pub(crate) fn postgres_default_expression(default_expression: &str, typ: &Type) -> Option { + parse_default_expression(default_expression, typ).map(|expression| match expression { + DefaultExpression::StringLiteral(value) + | DefaultExpression::NumericLiteral(value) + | DefaultExpression::BooleanLiteral(value) + | DefaultExpression::DateLiteral(value) + | DefaultExpression::TimeLiteral(value) + | DefaultExpression::TimeTzLiteral(value) + | DefaultExpression::TimestampLiteral(value) + | DefaultExpression::TimestampTzLiteral(value) + | DefaultExpression::IntervalLiteral(value) + | DefaultExpression::JsonLiteral(value) => value, + }) +} + +#[cfg(test)] +mod tests { + use etl::schema::{ + ColumnChange, ColumnModification, ColumnSchema, SchemaDiff, TableName, Type, + }; + + use super::{postgres_column_type_sql, schema_diff_statements}; + + #[test] + fn schema_diff_statements_orders_drop_rename_add_for_name_reuse() { + let destination_table = TableName::new("replica".to_owned(), "items".to_owned()); + let diff = SchemaDiff { + columns_to_add: vec![ColumnSchema::new("name".to_owned(), Type::TEXT, -1, 4, true)], + columns_to_remove: vec![ColumnSchema::new("value".to_owned(), Type::TEXT, -1, 3, true)], + columns_to_change: vec![ColumnChange { + ordinal_position: 2, + old_column: ColumnSchema::new("name".to_owned(), Type::TEXT, -1, 2, true), + new_column: ColumnSchema::new("value".to_owned(), Type::TEXT, -1, 2, true), + modifications: vec![ColumnModification::Rename { + old_name: "name".to_owned(), + new_name: "value".to_owned(), + }], + }], + }; + + let statements = schema_diff_statements(&destination_table, &diff); + assert_eq!(statements.len(), 3); + assert!( + statements[0].contains("drop column") && statements[0].contains("value"), + "expected drop of value first, got {:?}", + statements[0] + ); + assert!( + statements[1].contains("rename column") + && statements[1].contains("name") + && statements[1].contains("value"), + "expected rename name to value second, got {:?}", + statements[1] + ); + assert!( + statements[2].contains("add column") && statements[2].contains("name"), + "expected add of name third, got {:?}", + statements[2] + ); + assert!(!statements[2].contains("if not exists")); + } + + #[test] + fn postgres_column_type_sql_maps_timetz_to_text() { + assert_eq!(postgres_column_type_sql(&Type::TIMETZ, -1), "text"); + assert_eq!(postgres_column_type_sql(&Type::TIMETZ_ARRAY, -1), "text[]"); + } +} diff --git a/crates/etl-destinations/src/postgres/sql.rs b/crates/etl-destinations/src/postgres/sql.rs new file mode 100644 index 000000000..eb46ff369 --- /dev/null +++ b/crates/etl-destinations/src/postgres/sql.rs @@ -0,0 +1,14 @@ +//! SQL identifier helpers for the Postgres destination. + +use etl::schema::TableName; +use pg_escape::quote_identifier as pg_quote_identifier; + +/// Quotes a SQL identifier for safe inclusion in DDL/DML. +pub(crate) fn quote_identifier(identifier: &str) -> String { + pg_quote_identifier(identifier).to_string() +} + +/// Returns the quoted `"schema"."table"` form of [`TableName`]. +pub(crate) fn quote_table_name(table_name: &TableName) -> String { + table_name.as_quoted_identifier() +} diff --git a/crates/etl-destinations/src/postgres/test_utils.rs b/crates/etl-destinations/src/postgres/test_utils.rs new file mode 100644 index 000000000..cecff366a --- /dev/null +++ b/crates/etl-destinations/src/postgres/test_utils.rs @@ -0,0 +1,99 @@ +//! Test utilities for the Postgres destination. + +use etl::store::{SchemaStore, StateStore}; +use etl_config::shared::{PgConnectionConfig, TcpKeepaliveConfig}; +use etl_postgres::{test_utils::local_tls_config_from_env, tokio::test_utils::PgDatabase}; +use tokio_postgres::{Client, Row}; +use uuid::Uuid; + +use crate::{postgres::PostgresDestination, table_name::try_stringify_table_name}; +use etl::schema::TableName; + +/// Default schema override used by destination integration tests. +pub const TEST_DESTINATION_SCHEMA: &str = "replica"; + +/// Destination table name under the test schema override (`test_`). +pub fn destination_table_ident(source_table: &TableName) -> String { + try_stringify_table_name(source_table).expect("test table name should encode") +} + +/// Builds a destination database config on the same server as the test source. +pub fn destination_pg_connection_config() -> PgConnectionConfig { + PgConnectionConfig { + host: std::env::var("TESTS_DATABASE_HOST").unwrap_or_else(|_| "localhost".to_owned()), + hostaddr: None, + port: std::env::var("TESTS_DATABASE_PORT") + .unwrap_or_else(|_| "5430".to_owned()) + .parse() + .expect("TESTS_DATABASE_PORT must be a valid port number"), + name: format!("etl_pg_dest_{}", Uuid::new_v4().simple()), + username: std::env::var("TESTS_DATABASE_USERNAME") + .unwrap_or_else(|_| "postgres".to_owned()), + password: std::env::var("TESTS_DATABASE_PASSWORD") + .ok() + .or_else(|| Some("postgres".to_owned())) + .map(Into::into), + tls: local_tls_config_from_env(), + keepalive: TcpKeepaliveConfig::default(), + } +} + +/// Isolated Postgres database used as a destination in tests. +pub struct PostgresTestDatabase { + database: PgDatabase, +} + +impl PostgresTestDatabase { + /// Creates a unique destination database on the test Postgres server. + pub async fn setup() -> Self { + let config = destination_pg_connection_config(); + Self { database: PgDatabase::new(config).await } + } + + /// Returns the destination connection config. + pub fn config(&self) -> &PgConnectionConfig { + &self.database.config + } + + /// Builds a [`PostgresDestination`] scoped to this database. + pub fn build_destination(&self, store: S) -> PostgresDestination + where + S: StateStore + SchemaStore + Send + Sync, + { + PostgresDestination::new( + self.config().clone(), + Some(TEST_DESTINATION_SCHEMA.to_owned()), + store, + ) + } + + /// Runs a SQL statement against the destination database. + pub async fn run_sql(&self, sql: &str) { + self.database.run_sql(sql).await.expect("destination sql failed"); + } + + /// Returns raw query rows. + pub async fn query(&self, sql: &str) -> Vec { + let client = self.database.client.as_ref().expect("destination client missing"); + client.query(sql, &[]).await.expect("destination query failed") + } + + /// Returns column names for a destination table in ordinal order. + pub async fn column_names(&self, schema: &str, table: &str) -> Vec { + let client = self.database.client.as_ref().expect("destination client missing"); + let rows = client + .query( + "select column_name from information_schema.columns where table_schema = $1 and \ + table_name = $2 order by ordinal_position", + &[&schema, &table], + ) + .await + .expect("column lookup failed"); + rows.into_iter().map(|row| row.get(0)).collect() + } +} + +/// Sets up a destination database for tests. +pub async fn setup_postgres_destination_database() -> PostgresTestDatabase { + PostgresTestDatabase::setup().await +} diff --git a/crates/etl-destinations/tests/main.rs b/crates/etl-destinations/tests/main.rs index eaf8f1756..efa379f9d 100644 --- a/crates/etl-destinations/tests/main.rs +++ b/crates/etl-destinations/tests/main.rs @@ -8,5 +8,7 @@ mod clickhouse; mod ducklake; #[cfg(all(feature = "iceberg", feature = "test-utils"))] mod iceberg; +#[cfg(all(feature = "postgres", feature = "test-utils"))] +mod postgres; #[cfg(all(feature = "snowflake", feature = "test-utils"))] mod snowflake; diff --git a/crates/etl-destinations/tests/postgres/mod.rs b/crates/etl-destinations/tests/postgres/mod.rs new file mode 100644 index 000000000..eab2e1f41 --- /dev/null +++ b/crates/etl-destinations/tests/postgres/mod.rs @@ -0,0 +1 @@ +mod pipeline; diff --git a/crates/etl-destinations/tests/postgres/pipeline.rs b/crates/etl-destinations/tests/postgres/pipeline.rs new file mode 100644 index 000000000..0948b1951 --- /dev/null +++ b/crates/etl-destinations/tests/postgres/pipeline.rs @@ -0,0 +1,463 @@ +use etl::{ + event::EventType, + pipeline::PipelineId, + store::{StateStore, TableStateType}, + test_utils::{ + database::{spawn_source_database, test_table_name}, + event::EventCondition, + notifying_store::NotifyingStore, + pipeline::create_pipeline, + test_destination_wrapper::TestDestinationWrapper, + }, +}; +use etl_destinations::postgres::test_utils::{ + TEST_DESTINATION_SCHEMA, destination_table_ident, setup_postgres_destination_database, +}; +use etl_postgres::tokio::test_utils::TableModification; +use etl_telemetry::tracing::init_test_tracing; +use rand::random; + +use crate::support::crypto::install_crypto_provider; + +#[tokio::test(flavor = "multi_thread")] +async fn table_copy_roundtrip() { + init_test_tracing(); + install_crypto_provider(); + + let database = spawn_source_database().await; + let table_name = test_table_name("pg_copy_types"); + let table_id = database + .create_table( + table_name.clone(), + true, + &[ + ("smallint_col", "smallint not null"), + ("integer_col", "integer not null"), + ("bigint_col", "bigint not null"), + ("real_col", "real not null"), + ("double_col", "double precision not null"), + ("numeric_col", "numeric(10,2) not null"), + ("boolean_col", "boolean not null"), + ("text_col", "text not null"), + ("varchar_col", "varchar(100) not null"), + ("date_col", "date not null"), + ("timestamp_col", "timestamp not null"), + ("timestamptz_col", "timestamptz not null"), + ("uuid_col", "uuid not null"), + ("jsonb_col", "jsonb not null"), + ], + ) + .await + .expect("create table"); + + let publication_name = "test_pub_pg_copy"; + database + .create_publication(publication_name, std::slice::from_ref(&table_name)) + .await + .expect("create publication"); + + database + .run_sql(&format!( + r#"insert into {table} ( + smallint_col, integer_col, bigint_col, real_col, double_col, numeric_col, + boolean_col, text_col, varchar_col, date_col, timestamp_col, timestamptz_col, + uuid_col, jsonb_col + ) values ( + 42, 1000, 9999999, 1.5, 2.5, 12345.67, + true, 'hello text', 'hello varchar', '2024-01-15', '2024-01-15 12:00:00', + '2024-01-15 12:00:00+00', 'f47ac10b-58cc-4372-a567-0e02b2c3d479', + '{{"key":"value"}}' + )"#, + table = table_name.as_quoted_identifier(), + )) + .await + .expect("insert row"); + + let dest_db = setup_postgres_destination_database().await; + let store = NotifyingStore::new(); + let pipeline_id: PipelineId = random(); + let destination = dest_db.build_destination(store.clone()); + + let ready = store.notify_on_table_state_type(table_id, TableStateType::Ready).await; + let mut pipeline = create_pipeline( + &database.config, + pipeline_id, + publication_name.to_owned(), + store, + destination, + ); + pipeline.start().await.unwrap(); + ready.notified().await; + pipeline.shutdown_and_wait().await.unwrap(); + + let rows = dest_db + .query(&format!( + "select id, smallint_col, integer_col, bigint_col, text_col, varchar_col, \ + numeric_col::text, boolean_col, uuid_col::text from {schema}.{table} order by \ + id", + schema = TEST_DESTINATION_SCHEMA, + table = destination_table_ident(&table_name), + )) + .await; + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].get::<_, i64>(0), 1); + assert_eq!(rows[0].get::<_, i16>(1), 42); + assert_eq!(rows[0].get::<_, i32>(2), 1000); + assert_eq!(rows[0].get::<_, i64>(3), 9_999_999); + assert_eq!(rows[0].get::<_, String>(4), "hello text"); + assert_eq!(rows[0].get::<_, String>(5), "hello varchar"); + assert_eq!(rows[0].get::<_, String>(6), "12345.67"); + assert!(rows[0].get::<_, bool>(7)); + assert_eq!(rows[0].get::<_, String>(8).to_lowercase(), "f47ac10b-58cc-4372-a567-0e02b2c3d479"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn updates_and_deletes_streamed() { + init_test_tracing(); + install_crypto_provider(); + + let database = spawn_source_database().await; + let table_name = test_table_name("pg_update_delete"); + let table_id = database + .create_table(table_name.clone(), true, &[("value", "text not null")]) + .await + .expect("create table"); + + let publication_name = "test_pub_pg_upd_del"; + database + .create_publication(publication_name, std::slice::from_ref(&table_name)) + .await + .expect("create publication"); + + database + .run_sql(&format!( + "insert into {} (value) values ('keep'), ('drop')", + table_name.as_quoted_identifier(), + )) + .await + .expect("insert"); + + let dest_db = setup_postgres_destination_database().await; + let store = NotifyingStore::new(); + let destination = TestDestinationWrapper::wrap(dest_db.build_destination(store.clone())); + let ready = store.notify_on_table_state_type(table_id, TableStateType::Ready).await; + + let mut pipeline = create_pipeline( + &database.config, + random(), + publication_name.to_owned(), + store, + destination.clone(), + ); + pipeline.start().await.unwrap(); + ready.notified().await; + + let events = destination + .wait_for_events(vec![ + EventCondition::TableCount(EventType::Update, table_id, 1), + EventCondition::TableCount(EventType::Delete, table_id, 1), + ]) + .await; + + database + .run_sql(&format!( + "update {} set value = 'updated' where id = 1", + table_name.as_quoted_identifier(), + )) + .await + .expect("update"); + database + .run_sql(&format!("delete from {} where id = 2", table_name.as_quoted_identifier(),)) + .await + .expect("delete"); + events.notified().await; + pipeline.shutdown_and_wait().await.unwrap(); + + let rows = dest_db + .query(&format!( + "select id, value from {schema}.{table} order by id", + schema = TEST_DESTINATION_SCHEMA, + table = destination_table_ident(&table_name), + )) + .await; + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].get::<_, i64>(0), 1); + assert_eq!(rows[0].get::<_, String>(1), "updated"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn truncate_clears_table() { + init_test_tracing(); + install_crypto_provider(); + + let database = spawn_source_database().await; + let table_name = test_table_name("pg_truncate"); + let table_id = database + .create_table(table_name.clone(), true, &[("value", "text not null")]) + .await + .expect("create table"); + + let publication_name = "test_pub_pg_truncate"; + database + .create_publication(publication_name, std::slice::from_ref(&table_name)) + .await + .expect("create publication"); + database + .run_sql(&format!( + "insert into {} (value) values ('a'), ('b')", + table_name.as_quoted_identifier(), + )) + .await + .expect("insert"); + + let dest_db = setup_postgres_destination_database().await; + let store = NotifyingStore::new(); + let destination = TestDestinationWrapper::wrap(dest_db.build_destination(store.clone())); + let ready = store.notify_on_table_state_type(table_id, TableStateType::Ready).await; + + let mut pipeline = create_pipeline( + &database.config, + random(), + publication_name.to_owned(), + store, + destination.clone(), + ); + pipeline.start().await.unwrap(); + ready.notified().await; + + let events = destination + .wait_for_events(vec![ + EventCondition::TableCount(EventType::Truncate, table_id, 1), + EventCondition::TableCount(EventType::Insert, table_id, 1), + ]) + .await; + database + .run_sql(&format!("truncate {}", table_name.as_quoted_identifier())) + .await + .expect("truncate"); + database + .run_sql(&format!( + "insert into {} (value) values ('after')", + table_name.as_quoted_identifier(), + )) + .await + .expect("insert after truncate"); + events.notified().await; + pipeline.shutdown_and_wait().await.unwrap(); + + let rows = dest_db + .query(&format!( + "select id, value from {schema}.{table} order by id", + schema = TEST_DESTINATION_SCHEMA, + table = destination_table_ident(&table_name), + )) + .await; + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].get::<_, String>(1), "after"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn schema_change_add_drop_rename() { + init_test_tracing(); + install_crypto_provider(); + + let database = spawn_source_database().await; + let table_name = test_table_name("pg_schema_multi"); + let table_id = database + .create_table( + table_name.clone(), + true, + &[("name", "text not null"), ("age", "integer not null"), ("status", "text")], + ) + .await + .expect("create table"); + + let publication_name = "test_pub_pg_schema_multi"; + database + .create_publication(publication_name, std::slice::from_ref(&table_name)) + .await + .expect("create publication"); + database + .run_sql(&format!( + "insert into {} (name, age, status) values ('Alice', 25, 'active')", + table_name.as_quoted_identifier(), + )) + .await + .expect("insert"); + + let dest_db = setup_postgres_destination_database().await; + let store = NotifyingStore::new(); + let destination = TestDestinationWrapper::wrap(dest_db.build_destination(store.clone())); + let ready = store.notify_on_table_state_type(table_id, TableStateType::Ready).await; + + let mut pipeline = create_pipeline( + &database.config, + random(), + publication_name.to_owned(), + store.clone(), + destination.clone(), + ); + pipeline.start().await.unwrap(); + ready.notified().await; + + let initial_columns = + dest_db.column_names(TEST_DESTINATION_SCHEMA, &destination_table_ident(&table_name)).await; + assert_eq!(initial_columns, vec!["id", "name", "age", "status"]); + let initial_snapshot = store + .get_applied_destination_table_metadata(table_id) + .await + .unwrap() + .expect("metadata") + .snapshot_id; + + let events = destination + .wait_for_events(vec![ + EventCondition::TableCount(EventType::Relation, table_id, 1), + EventCondition::TableCount(EventType::Insert, table_id, 1), + ]) + .await; + + database + .alter_table( + table_name.clone(), + &[TableModification::RenameColumn { old_name: "name", new_name: "full_name" }], + ) + .await + .unwrap(); + database + .alter_table(table_name.clone(), &[TableModification::DropColumn { name: "age" }]) + .await + .unwrap(); + database + .alter_table( + table_name.clone(), + &[TableModification::AddColumn { name: "email", data_type: "text" }], + ) + .await + .unwrap(); + database + .run_sql(&format!( + "insert into {} (full_name, status, email) values ('Bob', 'pending', \ + 'bob@example.com')", + table_name.as_quoted_identifier(), + )) + .await + .expect("insert bob"); + events.notified().await; + pipeline.shutdown_and_wait().await.unwrap(); + + let final_columns = + dest_db.column_names(TEST_DESTINATION_SCHEMA, &destination_table_ident(&table_name)).await; + assert_eq!(final_columns, vec!["id", "full_name", "status", "email"]); + + let final_snapshot = store + .get_applied_destination_table_metadata(table_id) + .await + .unwrap() + .expect("metadata") + .snapshot_id; + assert!(final_snapshot > initial_snapshot); + + let rows = dest_db + .query(&format!( + "select id, full_name, status, email from {schema}.{table} order by id", + schema = TEST_DESTINATION_SCHEMA, + table = destination_table_ident(&table_name), + )) + .await; + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].get::<_, String>(1), "Alice"); + assert_eq!(rows[0].get::<_, Option>(3), None); + assert_eq!(rows[1].get::<_, String>(1), "Bob"); + assert_eq!(rows[1].get::<_, Option>(3), Some("bob@example.com".to_owned())); +} + +#[tokio::test(flavor = "multi_thread")] +async fn table_copy_reset_drops_destination() { + init_test_tracing(); + install_crypto_provider(); + + let database = spawn_source_database().await; + let table_name = test_table_name("pg_reset_copy"); + let table_id = database + .create_table(table_name.clone(), true, &[("value", "text not null")]) + .await + .expect("create table"); + + let publication_name = "test_pub_pg_reset_copy"; + database + .create_publication(publication_name, std::slice::from_ref(&table_name)) + .await + .expect("create publication"); + database + .run_sql(&format!( + "insert into {} (value) values ('old-1'), ('old-2')", + table_name.as_quoted_identifier(), + )) + .await + .expect("insert"); + + let dest_db = setup_postgres_destination_database().await; + let store = NotifyingStore::new(); + let pipeline_id: PipelineId = random(); + let destination = TestDestinationWrapper::wrap(dest_db.build_destination(store.clone())); + let ready = store.notify_on_table_state_type(table_id, TableStateType::Ready).await; + + let mut pipeline = create_pipeline( + &database.config, + pipeline_id, + publication_name.to_owned(), + store.clone(), + destination, + ); + pipeline.start().await.unwrap(); + ready.notified().await; + pipeline.shutdown_and_wait().await.unwrap(); + + let before = dest_db + .query(&format!( + "select count(*)::bigint from {schema}.{table}", + schema = TEST_DESTINATION_SCHEMA, + table = destination_table_ident(&table_name), + )) + .await; + assert_eq!(before[0].get::<_, i64>(0), 2); + + database + .run_sql(&format!("delete from {} where true", table_name.as_quoted_identifier())) + .await + .expect("delete source rows"); + database + .run_sql(&format!( + "insert into {} (value) values ('new-only')", + table_name.as_quoted_identifier(), + )) + .await + .expect("insert recopy row"); + + store.reset_table_state(table_id).await.expect("reset table state"); + + let destination = TestDestinationWrapper::wrap(dest_db.build_destination(store.clone())); + let ready = store.notify_on_table_state_type(table_id, TableStateType::Ready).await; + let mut pipeline = create_pipeline( + &database.config, + pipeline_id, + publication_name.to_owned(), + store.clone(), + destination.clone(), + ); + pipeline.start().await.unwrap(); + ready.notified().await; + pipeline.shutdown_and_wait().await.unwrap(); + + assert!(destination.was_table_dropped_for_copy(table_id).await); + let rows = dest_db + .query(&format!( + "select id, value from {schema}.{table} order by id", + schema = TEST_DESTINATION_SCHEMA, + table = destination_table_ident(&table_name), + )) + .await; + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].get::<_, String>(1), "new-only"); +} diff --git a/crates/etl-destinations/tests/support/mod.rs b/crates/etl-destinations/tests/support/mod.rs index 8e3cea7ed..c6c444fce 100644 --- a/crates/etl-destinations/tests/support/mod.rs +++ b/crates/etl-destinations/tests/support/mod.rs @@ -3,7 +3,8 @@ pub(crate) mod bigquery; pub(crate) mod clickhouse; #[cfg(any( all(feature = "bigquery", feature = "test-utils"), - all(feature = "clickhouse", feature = "test-utils") + all(feature = "clickhouse", feature = "test-utils"), + all(feature = "postgres", feature = "test-utils") ))] pub(crate) mod crypto; #[cfg(feature = "ducklake")] diff --git a/crates/etl-replicator/Cargo.toml b/crates/etl-replicator/Cargo.toml index 8d3f7a98d..76193fcdd 100644 --- a/crates/etl-replicator/Cargo.toml +++ b/crates/etl-replicator/Cargo.toml @@ -13,13 +13,14 @@ path = "src/bin/etl-ducklake-maintenance.rs" required-features = ["ducklake"] [features] -default = ["bigquery", "clickhouse", "ducklake", "iceberg", "snowflake"] +default = ["bigquery", "clickhouse", "ducklake", "iceberg", "snowflake", "postgres"] any-destination = [] bigquery = ["any-destination", "etl-destinations/bigquery"] clickhouse = ["any-destination", "etl-destinations/clickhouse"] ducklake = ["any-destination", "etl-destinations/ducklake", "etl-maintenance/ducklake"] iceberg = ["any-destination", "etl-destinations/iceberg"] snowflake = ["any-destination", "etl-destinations/snowflake"] +postgres = ["any-destination", "etl-destinations/postgres"] egress = ["etl/egress"] [dependencies] diff --git a/crates/etl-replicator/src/core/destinations.rs b/crates/etl-replicator/src/core/destinations.rs index 166ef0098..3ef635a7f 100644 --- a/crates/etl-replicator/src/core/destinations.rs +++ b/crates/etl-replicator/src/core/destinations.rs @@ -69,6 +69,17 @@ pub(super) async fn start( Err(disabled_destination_error(DestinationKind::Snowflake)) } } + DestinationKind::Postgres => { + #[cfg(feature = "postgres")] + { + postgres::start(replicator_config, store).await + } + + #[cfg(not(feature = "postgres"))] + { + Err(disabled_destination_error(DestinationKind::Postgres)) + } + } } } @@ -77,7 +88,8 @@ pub(super) async fn start( not(feature = "clickhouse"), not(feature = "ducklake"), not(feature = "iceberg"), - not(feature = "snowflake") + not(feature = "snowflake"), + not(feature = "postgres") ))] fn disabled_destination_error(kind: DestinationKind) -> crate::error::ReplicatorError { crate::error::ReplicatorError::config(std::io::Error::other(format!( @@ -415,3 +427,35 @@ mod snowflake { pipeline::start(pipeline).await } } + +/// Postgres destination startup. +#[cfg(feature = "postgres")] +mod postgres { + use etl::pipeline::Pipeline; + use etl_config::shared::{DestinationConfig, ReplicatorConfig}; + use etl_destinations::postgres::PostgresDestination; + + use super::super::{ReplicatorStore, pipeline}; + use crate::error::ReplicatorResult; + + /// Starts the Postgres destination pipeline. + pub(super) async fn start( + replicator_config: ReplicatorConfig, + store: ReplicatorStore, + ) -> ReplicatorResult<()> { + let DestinationConfig::Postgres { pg_connection, destination_schema } = + &replicator_config.destination + else { + unreachable!("Destination kind should match Postgres config"); + }; + + let destination = PostgresDestination::new( + pg_connection.clone(), + destination_schema.clone(), + store.clone(), + ); + + let pipeline = Pipeline::new(replicator_config.pipeline, store, destination); + pipeline::start(pipeline).await + } +} diff --git a/docs/src/content/docs/explanation/schema-changes.md b/docs/src/content/docs/explanation/schema-changes.md index 42cb58dac..2b19509a6 100644 --- a/docs/src/content/docs/explanation/schema-changes.md +++ b/docs/src/content/docs/explanation/schema-changes.md @@ -10,7 +10,7 @@ current implementation is intentionally conservative: the source-side event trigger captures a rich PostgreSQL-shaped snapshot, while ETL currently models well-understood column changes: **adds, drops, renames, and column default changes**. Built-in destination support varies by destination DDL -capabilities. **BigQuery, ClickHouse, DuckLake, and Snowflake** apply supported +capabilities. **BigQuery, ClickHouse, DuckLake, Snowflake, and Postgres** apply supported schema changes automatically; Iceberg is deprecated for new deployments and does not support schema-change DDL. @@ -75,6 +75,7 @@ ETL has one shared schema-change signal, but **DDL behavior is implemented per d |-------------|----------------------| | BigQuery | Supports add, drop, rename, `REQUIRED` to `NULLABLE` relaxation, and supported literal default metadata. BigQuery requires added columns to be nullable and does not backfill existing rows for `ADD COLUMN ... DEFAULT`. PostgreSQL remains responsible for enforcing later `SET NOT NULL` changes because BigQuery cannot tighten an existing column in place. | | ClickHouse | Supports add, drop, rename, and supported literal defaults. `ReplacingMergeTree` rejects primary-key drops or renames because the ordering expression cannot be rewritten safely. ClickHouse default expressions are metadata-only unless explicitly materialized; ETL does not issue `MATERIALIZE COLUMN`. Relation events whose schema snapshot is older than the applied destination snapshot are rejected instead of executing reverse DDL; recovering from that state requires resynchronizing the table. | +| Postgres | Supports add, drop, rename, DROP NOT NULL, and supported literal defaults. SET NOT NULL on existing columns is skipped with a warning. Tables are current-state UPSERT tables without CDC meta columns. Schema DDL is applied as DROP, then RENAME, then ADD, then nullability/default changes. Source `timetz` / `timetz[]` columns are created as `text` / `text[]` so binds match the string encoding used for those values. When `destination_schema` is set, destination table names encode the source schema (`public.users` → `.public_users`) to avoid collisions. | | DuckLake | Supports add, drop, rename, and supported literal defaults. DuckLake records supported add-time defaults as metadata without rewriting existing data files. | | Snowflake | Supports add, drop, rename, create-table literal defaults, and literal add-column defaults. Literal defaults are included in `ADD COLUMN` so Snowflake can expose add-time default values for existing rows; non-literal defaults and later default changes are skipped with a warning. Relation events whose schema snapshot is older than the applied destination snapshot are skipped because Snowflake's durable channel offset safely deduplicates their replayed row events. | | Iceberg | Deprecated for now. Schema-change DDL is not a supported path for new deployments. | @@ -181,9 +182,15 @@ Destination support may be narrower than parser support: |-------------|----------------------------| | BigQuery | Supports compatible string, numeric, boolean, date, time, timestamp, JSON, and UUID literals. Added columns are created nullable and supported defaults are set afterward for future writes. | | ClickHouse | Supports compatible string, numeric, boolean, date, time, timestamp, JSON, and UUID literals. Defaults are metadata only unless separately materialized. | +| Postgres | Supports the same portable literal defaults as the source Postgres expressions ETL can parse. | | DuckLake | Supports compatible string, numeric, date, time, timestamp, JSON, and UUID literals. Boolean defaults are currently skipped by the DuckLake destination. | | Snowflake | `CREATE TABLE` supports compatible string, numeric, boolean, date, time, timestamp, JSON, and UUID literals. `ADD COLUMN` only receives the literal subset Snowflake allows for add-column defaults: string, numeric, and boolean literals. Later default changes on existing columns are skipped. | +API-created Postgres destinations currently force TLS disabled +(`TlsConfig::disabled()`). Library and replicator configs can still enable TLS +through `PgConnectionConfig.tls`. Source `timetz` values are stored in +destination `text` columns for reliable parameterized binds. + When changing a default from one supported expression to an unsupported expression, destinations that can safely remove defaults drop the old supported default to avoid leaving stale destination behavior behind. Snowflake is the @@ -227,7 +234,7 @@ A practical flow is: actually ready for following row events. 7. Process following row events with the new schema. -The built-in BigQuery, ClickHouse, DuckLake, and Snowflake destinations follow +The built-in BigQuery, ClickHouse, DuckLake, Snowflake, and Postgres destinations follow this shape: they mark destination schema metadata as `Applying`, apply the supported DDL operations, then mark the schema as `Applied`. Because destination DDL is not always transactional, a crash while metadata is `Applying` may require @@ -290,7 +297,7 @@ These behaviors are **not full destination DDL semantics** yet: - `ADD COLUMN ... DEFAULT` semantics differ by destination. ETL intentionally avoids destination DDL that rewrites all existing rows. BigQuery leaves pre-existing destination rows null for newly added defaulted columns, while - ClickHouse, DuckLake, and Snowflake can expose supported add-time defaults + ClickHouse, DuckLake, Snowflake, and Postgres can expose supported add-time defaults without ETL issuing a materialization rewrite. Snowflake only receives add-column defaults for source defaults that can be rendered as Snowflake literals.