This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12528-da28cf714b80d375697979b001f1b29373145f18 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 9338f98e306116ef0f57c334efef7336534036f5 Author: yzeng1618 <[email protected]> AuthorDate: Thu Oct 22 15:41:31 2026 +0000 [Fix][Connector-V2] Stream PostgreSQL sampling results with JDBC cursor (#12731) --- .../seatunnel/cdc/postgres/utils/PostgresUtils.java | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresUtils.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresUtils.java index dda4a4c2e0..6067abeae9 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresUtils.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresUtils.java @@ -179,13 +179,17 @@ public class PostgresUtils { Statement stmt = null; ResultSet rs = null; + Connection connection = jdbc.connection(); List<Object> results = new ArrayList<>(); + // pgjdbc honors fetchSize only in cursor mode, which requires auto-commit off; otherwise + // the driver silently materializes the whole column in this JVM before the first next(). + boolean previousAutoCommit = connection.getAutoCommit(); + connection.setAutoCommit(false); try { stmt = - jdbc.connection() - .createStatement( - ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + connection.createStatement( + ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(DEFAULT_FETCH_SIZE); rs = stmt.executeQuery(sampleQuery); @@ -218,6 +222,12 @@ public class PostgresUtils { log.error("Failed to close Statement", e); } } + try { + connection.setAutoCommit(previousAutoCommit); + } catch (SQLException e) { + // Preserve the original sampling exception if restoring the connection state fails. + log.warn("Failed to restore auto-commit state", e); + } } Object[] resultsArray = results.toArray(); Arrays.sort(resultsArray);
