fix: postgres money null bug

fix: starrocks update partial columns bug

Author: jarad0628

connectors/postgres-connector/src/main/java/io/tapdata/connector/postgres/PostgresConnector.java

@@ -1005,7 +1005,7 @@ private Map<String, Object> filterTimeForPG(ResultSet resultSet, Map<String, Str
 
                 } else if (dataType.equals("money")) {
                     String money = resultSet.getString(colName);
-                    if ("null".equals(money)) {
+                    if (EmptyKit.isBlank(money) || "null".equals(money)) {
                         dataMap.put(colName, null);
                     } else {
                         dataMap.put(colName, new BigDecimal(money.replaceAll("[^\\d.-]", "")));

connectors/starrocks-connector/src/main/java/io/tapdata/connector/starrocks/StarrocksConnector.java

@@ -58,6 +58,7 @@ public class StarrocksConnector extends CommonDbConnector {
     private StarrocksConfig starrocksConfig;
     private final Map<String, StarrocksStreamLoader> starrocksStreamLoaderMap = new ConcurrentHashMap<>();
     private Consumer<Object> flushOffsetCallback;
+    private boolean hasMethodError = false;
 
 
     @Override
@@ -181,6 +182,7 @@ protected RetryOptions errorHandle(TapConnectionContext tapConnectionContext, PD
             retryOptions.needRetry(true);
             return retryOptions;
         }
+        hasMethodError = true;
         return retryOptions;
     }
 
@@ -196,6 +198,19 @@ public StarrocksStreamLoader getStarrocksStreamLoader() {
     }
 
     private void writeRecord(TapConnectorContext connectorContext, List<TapRecordEvent> tapRecordEvents, TapTable tapTable, Consumer<WriteListResult<TapRecordEvent>> writeListResultConsumer) throws Throwable {
+        if (hasMethodError) {
+            for (StarrocksStreamLoader starrocksStreamLoader : starrocksStreamLoaderMap.values()) {
+                if (EmptyKit.isNotNull(starrocksStreamLoader)) {
+                    try {
+                        starrocksStreamLoader.flushOnStop();
+                        tapLogger.info("StarrocksConnector", "Flushed remaining data before stopping StarrocksStreamLoader");
+                    } catch (Exception e) {
+                        tapLogger.warn("StarrocksConnector", "Failed to flush data before stopping: {}", e.getMessage());
+                    }
+                }
+            }
+            hasMethodError = false;
+        }
         try {
             if (checkStreamLoad()) {
                 getStarrocksStreamLoader().writeRecord(tapRecordEvents, tapTable, writeListResultConsumer);

connectors/starrocks-connector/src/main/java/io/tapdata/connector/starrocks/streamload/StarrocksStreamLoader.java

@@ -963,19 +963,13 @@ public RespContent flushTable(String tableName, TapTable table) throws Starrocks
             taplogger.warn("Table {} flush failed: flushed_size={}, waiting_time={} ms, " +
                 "flush_duration={} ms, error={}",
                 tableName, formatBytes(tableDataSize), waitTime, flushDuration, e.getMessage());
-
-            // 失败：保留缓存文件（不删除），仅关闭文件流
-            cleanupCacheFileForTable(tableName, false);
             throw e;
         } catch (Exception e) {
             long flushEndTime = System.currentTimeMillis();
             long flushDuration = flushEndTime - flushStartTime;
             taplogger.warn("Table {} flush failed: flushed_size={}, waiting_time={} ms, " +
                 "flush_duration={} ms, error={}",
                 tableName, formatBytes(tableDataSize), waitTime, flushDuration, e.getMessage());
-
-            // 失败：保留缓存文件（不删除），仅关闭文件流
-            cleanupCacheFileForTable(tableName, false);
             throw new StarrocksRuntimeException(e);
         } finally {
 
@@ -1275,7 +1269,7 @@ private void cleanupCacheFileForTable(String tableName, boolean deleteFile) {
 
             // 清理相关状态
             isFirstRecordByTable.remove(tableName);
-//            dataColumnsByTable.remove(tableName);
+            dataColumnsByTable.remove(tableName);
             currentBatchSizeByTable.remove(tableName);
             // 注意：lastFlushTimeByTable 不清理，需要保持刷新时间记录
             // 注意：tableNameToTapTableMap 不清理，因为表结构信息需要持久保存