{"record":{"id":"660253080d07b47a","repo":"risingwavelabs/risingwave","slug":"max-batch-rows-must-be-between-1-and-got","errorCode":null,"errorMessage":"`max_batch_rows` must be between 1 and {}, got {}","messagePattern":"`max_batch_rows` must be between 1 and (.+?), got (.+?)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/postgres.rs","lineNumber":234,"sourceCode":"\n    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {\n        let schema = param.schema();\n        let pk_indices = param.downstream_pk_or_empty();\n        let config = PostgresConfig::from_btreemap(param.properties)?;\n        PostgresSink::new(config, schema, pk_indices, param.sink_type.is_append_only())\n    }\n}\n\nimpl Sink for PostgresSink {\n    type LogSinker = BatchingLogSinker<PostgresSinkWriter>;\n\n    const SINK_NAME: &'static str = POSTGRES_SINK;\n\n    crate::impl_validate_sink_unknown_fields!();\n\n    async fn validate(&self) -> Result<()> {\n        if !(1..=MAX_BATCH_ROWS_LIMIT).contains(&self.config.max_batch_rows) {\n            return Err(SinkError::Config(anyhow!(\n                \"`max_batch_rows` must be between 1 and {}, got {}\",\n                MAX_BATCH_ROWS_LIMIT,\n                self.config.max_batch_rows\n            )));\n        }\n\n        if !self.is_append_only && self.pk_indices.is_empty() {\n            return Err(SinkError::Config(anyhow!(\n                \"Primary key not defined for upsert Postgres sink (please define in `primary_key` field)\"\n            )));\n        }\n\n        ensure_no_foreign_key(&self.config).await?;\n\n        // Verify our sink schema is compatible with Postgres\n        {\n            let pg_conn = self.config.pg_connection_config();\n            let pg_table = PostgresExternalTable::connect(","sourceCodeStart":216,"sourceCodeEnd":252,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/postgres.rs#L216-L252","documentation":"The PostgreSQL sink validates that the `max_batch_rows` sink config option is between 1 and MAX_BATCH_ROWS_LIMIT. `validate()` runs when the sink is created (CREATE SINK). If the user supplied a value outside that inclusive range, the sink refuses to start with this config error.","triggerScenarios":"CREATE SINK ... WITH (connector='postgres', max_batch_rows = 0) or max_batch_rows set to a value greater than MAX_BATCH_ROWS_LIMIT; also a negative value.","commonSituations":"Copy-pasted config with max_batch_rows = 0 (thinking 0 means 'unlimited batching'); typos producing huge numbers; tuning batch size for throughput and overshooting the compile-time limit.","solutions":["Set `max_batch_rows` to a value between 1 and MAX_BATCH_ROWS_LIMIT in the WITH options of CREATE SINK.","Remove the `max_batch_rows` option entirely to fall back to the default batch size.","Check the current limit constant in src/connector/src/sink/postgres.rs before tuning."],"exampleFix":"// before\nWITH (connector = 'postgres', max_batch_rows = 0)\n// after\nWITH (connector = 'postgres', max_batch_rows = 1024)","handlingStrategy":"validation","validationCode":"let max_batch_rows: usize = 1024;\nassert!((1..=4294967295).contains(&max_batch_rows), \"max_batch_rows must be between 1 and MAX_BATCH_ROWS_LIMIT\");","typeGuard":"fn valid_batch_rows(v: i64) -> bool { v >= 1 }","tryCatchPattern":null,"preventionTips":["Clamp max_batch_rows to >= 1 when generating sink configs.","Never use 0 to mean 'unbounded' for this option.","Consult MAX_BATCH_ROWS_LIMIT in the source before tuning."],"tags":["rust","postgres","sink","config"],"backgroundTag":"value-out-of-range","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}