Flink PostgreSQL JDBC Sink批量插入上限128，reWriteBatchedInserts未达预期

环境与场景 #

• Flink版本：1.16.2
• Timescale版本：2.11.0
• PostgreSQL版本：15.3
• 连接URL已配置reWriteBatchedInserts=true，使用Flink JdbcSink向Timescale超表写入数据，Sink代码中设置了withBatchSize(1000)。

问题现象 #

预期每次批量插入最多处理1000行数据，但实际每次插入不超过128行：

• 当withBatchSize(N)的N≤128时，符合预期；
• 当N>128时，实际效果等同于N=128；
• PostgreSQL全量查询日志显示，每次执行的插入语句固定为128行，对应896个绑定参数（128×7）。

Sink代码示例 #

SinkFunction<EnrichedDataTuple> dataSink = JdbcSink.sink(
                "insert into mytable (a, b, c, d, e, f, g) values (?,?,?,?,?,?,?) on conflict do nothing",
                (statement, row) -> { 
                    statement.setInt(1, row.getA());
                    statement.setInt(2, row.getB());
                    statement.setInt(3, row.getC());
                    statement.setInt(4, row.getD());
                    statement.setInt(5, row.getE());
                    statement.setInt(6, row.getF());
                    statement.setInt(7, row.getG());
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(1000)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(appConfiguration.getString(ConfigOptions.key("timescale.uri").stringType().noDefaultValue()))
                        .withUsername(appConfiguration.getString(ConfigOptions.key("timescale.user").stringType().noDefaultValue()))
                        .withPassword(appConfiguration.getString(ConfigOptions.key("timescale.password").stringType().noDefaultValue()))
                        .withDriverName("org.postgresql.Driver")
                        .build()
                );

实际执行的SQL示例 #

insert into public.mytable (a, b, c, d, e, f, g) values ($1,$2,$3,$4,$5,$6,$7),  
($8,$9,$10,$11,$12,$13,$14),($15,$16,$17,$18,$19,$20,$21),  
($22,$23,$24,$25,$26,$27,$28),($29,$30,$31,$32,$33,$34,$35),  
...
($883,$884,$885,$886,$887,$888,$889),($890,$891,$892,$893,$894,$895,$896) on conflict do nothing

问题根源 #

pgjdbc驱动的PgPreparedStatement.java类中存在硬编码限制：

@Override
protected void transformQueriesAndParameters() throws SQLException {
    ArrayList<@Nullable ParameterList> batchParameters = this.batchParameters;
    if (batchParameters == null || batchParameters.size() <= 1
        || !(preparedQuery.query instanceof BatchedQuery)) {
      return;
    }
    BatchedQuery originalQuery = (BatchedQuery) preparedQuery.query;
    // Single query cannot have more than {@link Short#MAX_VALUE} binds, thus
    // the number of multi-values blocks should be capped.
    // Typically, it does not make much sense to batch more than 128 rows: performance
    // does not improve much after updating 128 statements with 1 multi-valued one, thus
    // we cap maximum batch size and split there.
    final int bindCount = originalQuery.getBindCount();
    final int highestBlockCount = 128;
    final int maxValueBlocks = bindCount == 0 ? 1024 /* if no binds, use 1024 rows */
        : Integer.highestOneBit( // deriveForMultiBatch supports powers of two only
            Math.min(Math.max(1, maximumNumberOfParameters() / bindCount), highestBlockCount));
}

驱动开发者基于性能考量，硬编码了highestBlockCount = 128，限制了单批次转换后的行数。

解决方案与疑问解答 #

1. 是否有PostgreSQL参数或JDBC选项可调？ #

目前pgjdbc驱动没有提供官方配置项修改这个128行的硬限制，无法通过连接参数或数据库配置直接调整。

2. 重新编译驱动增大此限制是否可行？ #

技术上完全可行，但需要注意以下几点：

• 参数上限约束：PostgreSQL单语句的绑定参数总数不能超过Short.MAX_VALUE（32767），结合每行7个参数，最大可支持的行数为32767 / 7 ≈ 4681，调整时不要超过这个范围，设置1000行是符合要求的。
• 性能验证：官方注释提到超过128行性能提升不明显，但实际效果可能因业务场景（如数据量、数据库负载、网络延迟）而异，需要自行测试不同批量大小对插入吞吐量和数据库压力的影响。
• 维护成本：自定义编译驱动后，后续升级pgjdbc版本时需要同步修改该硬编码值，同时要确保自定义驱动与Flink、PostgreSQL的版本兼容性。

3. 替代方案 #

如果不想自定义驱动，可以考虑：

• 保持withBatchSize(128)，结合调整withBatchIntervalMs参数，控制批次触发的时间间隔，平衡吞吐量和延迟；
• 自定义Sink实现，使用PostgreSQL的COPY命令批量写入，Timescale对COPY的支持较好，吞吐量通常高于批量INSERT，但需要额外开发Sink逻辑。

内容的提问来源于stack exchange，提问作者LucaD