Dev

CHANGELOG.md

@@ -33,6 +33,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Fixed
 
+- Data task streaming memory and completeness: the single statement now executes
+  in an explicit transaction (auto-commit restored when handing the connection
+  back, DML committed first) so PostgreSQL-family drivers — including Hologres —
+  honor `fetch-size` and pull rows in bounded batches instead of buffering the
+  whole result set in heap, which caused `Java heap space` / repeated
+  `GC overhead limit exceeded` failures on large exports. Also fixed a dropped
+  trailing partial batch (fewer than 1000 rows) whenever a result set ended
+  normally; both are locked by `DataTaskJobEngineStreamQueryTest`.
 - Deployment timezone alignment: the compose MySQL now starts with
   `--default-time-zone=+08:00` (`build-docker/install`, `.devcontainer`) and the devcontainer
   PostgreSQL sets `timezone=Asia/Shanghai`, matching the JDBC `serverTimezone=Asia/Shanghai`

build-docker/datapoly/datapoly-release/bin/datapolyctl.sh

@@ -21,10 +21,13 @@ echo "Base Directory:${APP_HOME}"
 export APP_DRIVERS_PATH=$APP_HOME/drivers
 
 # JVM参数可以在这里设置
+# 堆 4G、年轻代/老年代 1:3：长驻对象（Hazelcast token/API 响应缓存、Eureka、Spring 框架
+# 对象）占堆内大头，老年代空间优先；年轻代 1G 足以容纳数据任务流式批次与 ≤200 行的
+# 调试/预览结果集（线上观测：16 分钟仅 7 次 minor GC，老年代却 30 秒内 5→25 次 major GC）。
 # -XX:+PerfDisableSharedMem: the JDK perfdata file lands in java.io.tmpdir; when the host
 # bind-mounts /tmp (macOS Docker Desktop virtiofs), zeroing that 32KB mmap SIGBUSes the JVM
 # at startup. Disabling shared-mem perfdata removes the mmap entirely.
-JVMFLAGS="-server -Xms1024m -Xmx1024m -Xmn1024m -XX:+DisableExplicitGC -XX:+PerfDisableSharedMem -Djava.awt.headless=true -Dfile.encoding=UTF-8 "
+JVMFLAGS="-server -Xms4096m -Xmx4096m -Xmn1024m -XX:+DisableExplicitGC -XX:+PerfDisableSharedMem -Djava.awt.headless=true -Dfile.encoding=UTF-8 "
 
 if [ "$JAVA_HOME" != "" ]; then
   JAVA="$JAVA_HOME/bin/java"

datapoly-core/src/main/java/com/cs/core/datatask/DataTaskJobEngine.java

@@ -267,10 +267,20 @@ public Map<String, Object> debugPreview(DataTaskDefEntity def, Map<String, Objec
      */
     protected StreamResult streamQuery(StreamSpec spec, ResultChannel channel) throws Exception {
         Connection connection = spec.getDataSource().getConnection();
+        boolean autoCommitChanged = false;
         try {
             Consumer<Connection> executeBeforeQuery = spec.getProduct().getContext().getExecuteBeforeQuery();
             LambdaUtils.ifDo(null != executeBeforeQuery, () -> executeBeforeQuery.accept(connection));
 
+            // PostgreSQL-family drivers (including Hologres) ignore the ResultSet fetch
+            // size while autocommit is on and buffer the whole result set in heap. Move
+            // the single statement into an explicit transaction so setFetchSize below
+            // bounds every network fetch to that many rows.
+            if (connection.getAutoCommit()) {
+                connection.setAutoCommit(false);
+                autoCommitChanged = true;
+            }
+
             PreparedStatement statement = connection.prepareStatement(spec.getSqlMeta().getSql());
             try {
                 statement.setQueryTimeout(spec.getTimeoutSeconds());
@@ -281,13 +291,23 @@ protected StreamResult streamQuery(StreamSpec spec, ResultChannel channel) throw
                 }
                 boolean hasResult = statement.execute();
                 if (!hasResult) {
+                    // DML ran outside autocommit: commit it, otherwise handing the
+                    // connection back rolls the transaction (and the rows) away.
+                    connection.commit();
                     return StreamResult.update(statement.getUpdateCount());
                 }
                 return drainResultSet(statement.getResultSet(), spec, channel);
             } finally {
                 statement.close();
             }
         } finally {
+            if (autoCommitChanged) {
+                try {
+                    connection.setAutoCommit(true);
+                } catch (SQLException e) {
+                    // the connection is about to be closed; nothing left to salvage
+                }
+            }
             connection.close();
         }
     }
@@ -338,8 +358,10 @@ private StreamResult drainResultSet(ResultSet rs, StreamSpec spec, ResultChannel
                     }
                 }
             }
-            if (!reachedEnd && !stoppedByChannel && !truncated && !buffer.isEmpty()) {
-                // trailing partial batch smaller than WRITE_BATCH_ROWS
+            if (!stoppedByChannel && !truncated && !buffer.isEmpty()) {
+                // trailing partial batch smaller than WRITE_BATCH_ROWS: this also fires
+                // when the result set ends normally, otherwise the final rows < batch
+                // size are silently dropped from the delivery
                 if (channel.batch(buffer)) {
                     tickProgress(spec, total, lastFlush);
                 }