Merge branch 'develop'

Author: jarad0628

connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/PostgresTest.java

@@ -245,7 +245,7 @@ private boolean deployPgto() {
             process = processBuilder.start();
             process.waitFor(60, TimeUnit.SECONDS);
             new File(FileUtil.paths(toolPath, ctlDir)).mkdir();
-            processBuilder.command("/bin/sh", "-c", String.format("walminer pgto -i -c %s -s '%s' -e %s -t 4 --source-connstr1='host=%s port=%s username=%s dbanme=%s password=%s'", toolDir.getAbsolutePath() + "/" + ctlDir, slotName, ((PostgresConfig) commonDbConfig).getPgtoPort(), commonDbConfig.getHost(), commonDbConfig.getPort(), commonDbConfig.getUser(), commonDbConfig.getDatabase(), commonDbConfig.getPassword()));
+            processBuilder.command("/bin/sh", "-c", String.format("walminer pgto init -c %s -s '%s' -e %s -t 4 --source-connstr1='host=%s port=%s username=%s dbanme=%s password=%s'", toolDir.getAbsolutePath() + "/" + ctlDir, slotName, ((PostgresConfig) commonDbConfig).getPgtoPort(), commonDbConfig.getHost(), commonDbConfig.getPort(), commonDbConfig.getUser(), commonDbConfig.getDatabase(), commonDbConfig.getPassword()));
             process = processBuilder.start();
             process.waitFor(60, TimeUnit.SECONDS);
             try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
@@ -256,7 +256,7 @@ private boolean deployPgto() {
             }
             int exitCode = process.waitFor();
             System.out.println("exit code: " + exitCode);
-            processBuilder.command("/bin/sh", "-c", String.format("walminer pgto -m -c %s", toolDir.getAbsolutePath() + "/" + ctlDir));
+            processBuilder.command("/bin/sh", "-c", String.format("walminer pgto run -m -c %s", toolDir.getAbsolutePath() + "/" + ctlDir));
             processBuilder.redirectOutput(ProcessBuilder.Redirect.INHERIT);
             process = processBuilder.start();
             process.waitFor(60, TimeUnit.SECONDS);

connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/AbstractWalLogMiner.java

@@ -176,6 +176,7 @@ protected void parseKeyAndValue(String tableName, Map.Entry<String, Object> stri
             case "integer":
             case "bigint":
             case "numeric":
+            case "money":
             case "real":
             case "double precision":
                 stringObjectEntry.setValue(new BigDecimal((String) value));

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

@@ -333,16 +333,16 @@ protected TapField makeTapField(DataMap dataMap) {
     @Override
     public void onStop(TapConnectionContext connectionContext) {
         ErrorKit.ignoreAnyError(() -> {
-            for (StarrocksStreamLoader StarrocksStreamLoader : starrocksStreamLoaderMap.values()) {
-                if (EmptyKit.isNotNull(StarrocksStreamLoader)) {
+            for (StarrocksStreamLoader starrocksStreamLoader : starrocksStreamLoaderMap.values()) {
+                if (EmptyKit.isNotNull(starrocksStreamLoader)) {
                     // 在停止前先刷新剩余数据
                     try {
-                        StarrocksStreamLoader.flushOnStop();
+                        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());
                     }
-                    StarrocksStreamLoader.shutdown();
+                    starrocksStreamLoader.shutdown();
                 }
             }
         });
@@ -367,6 +367,22 @@ protected void fieldDDLHandler(TapConnectorContext tapConnectorContext, TapField
         if (null == sqlList) {
             return;
         }
+        //执行ddl前需要将缓存的数据先flush，清空StarrocksStreamLoaderMap对象
+        ErrorKit.ignoreAnyError(() -> {
+            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());
+                    }
+                    starrocksStreamLoader.shutdown();
+                }
+            }
+            starrocksStreamLoaderMap.clear();
+        });
         try {
             tapLogger.info("Field ddl sql: {}", sqlList);
             jdbcContext.batchExecute(sqlList);