Repository: risingwavelabs/risingwave
File: src/connector/src/sink/postgres.rs (Rust)

Code before:
    #[serde(default = "Default::default")]
    pub ssl_mode: SslMode,
    #[serde(rename = "ssl.root.cert")]
    pub ssl_root_cert: Option<String>,
    #[serde(default = "default_max_batch_rows")]
    #[serde_as(as = "DisplayFromStr")]
    pub max_batch_rows: usize,
    pub r#type: String, // accept "append-only" or "upsert"
    #[serde(default, rename = "tcp.keepalive.enable")]
    #[serde_as(as = "DisplayFromStr")]
    pub tcp_keepalive_enable: bool,
    #[serde(default = "default_tcp_keepalive_idle", rename = "tcp.keepalive.idle")]
    #[serde_as(as = "DisplayFromStr")]
    pub tcp_keepalive_idle: u32,
    #[serde(
        default = "default_tcp_keepalive_interval",
        rename = "tcp.keepalive.interval"
    )]
    #[serde_as(as = "DisplayFromStr")]
    pub tcp_keepalive_interval: u32,
    #[serde(
        default = "default_tcp_keepalive_count",
        rename = "tcp.keepalive.count"
    )]
    #[serde_as(as = "DisplayFromStr")]
    pub tcp_keepalive_count: u32,
}

impl EnforceSecret for PostgresConfig {
    const ENFORCE_SECRET_PROPERTIES: phf::Set<&'static str> = phf_set! {
        "password", "ssl.root.cert"
    };
}

fn default_max_batch_rows() -> usize {
    1024
}

fn default_schema() -> String {
    "public".to_owned()
}

fn default_tcp_keepalive_idle() -> u32 {
    10 * 60 // 10 minutes
}

fn default_tcp_keepalive_interval() -> u32 {
    10
}

fn default_tcp_keepalive_count() -> u32 {

Review comment (suggestion):
The TCP keep-alive configuration parameters lack input validation in Rust. Consider adding validation to ensure values are within reasonable ranges (e.g., idle time > 0, interval > 0, count > 0). Invalid values could lead to unexpected behavior or connection issues.

Code after:
    #[serde(rename = "ssl.root.cert")]
    pub ssl_root_cert: Option<String>,
    #[serde(default = "default_max_batch_rows")]
    #[serde_as(as = "DisplayFromStr")]
    pub max_batch_rows: usize,
    pub r#type: String, // accept "append-only" or "upsert"
    #[serde(default, rename = "tcp.keepalive.enable")]
    #[serde_as(as = "DisplayFromStr")]
    pub tcp_keepalive_enable: bool,

    #[serde(flatten)]
    pub tcp_keepalive: Option<TcpKeepaliveConfig>,
}

#[serde_as]
#[derive(Debug, Clone, Deserialize)]
pub struct TcpKeepaliveConfig {
    #[serde(rename = "tcp.keepalive.idle")]
    #[serde_as(as = "DisplayFromStr")]
    pub tcp_keepalive_idle: u32,
    #[serde(rename = "tcp.keepalive.interval")]
    #[serde_as(as = "DisplayFromStr")]
    pub tcp_keepalive_interval: u32,
    #[serde(rename = "tcp.keepalive.count")]
    #[serde_as(as = "DisplayFromStr")]
    pub tcp_keepalive_count: u32,
}

impl Default for TcpKeepaliveConfig {
    fn default() -> Self {
        Self {
            tcp_keepalive_idle: 10 * 60, // 10 minutes,
            tcp_keepalive_interval: 10,
            tcp_keepalive_count: 3,
        }
    }
}

impl EnforceSecret for PostgresConfig {
    const ENFORCE_SECRET_PROPERTIES: phf::Set<&'static str> = phf_set! {
        "password", "ssl.root.cert"
    };
}

fn default_max_batch_rows() -> usize {
    1024
}

fn default_schema() -> String {
    "public".to_owned()
}