feat: TAP-6310

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

@@ -2,6 +2,7 @@
 
 import io.netty.buffer.ByteBufInputStream;
 import io.tapdata.common.dml.NormalWriteRecorder;
+import io.tapdata.entity.event.dml.TapDeleteRecordEvent;
 import io.tapdata.entity.event.dml.TapRecordEvent;
 import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
 import io.tapdata.entity.schema.TapTable;
@@ -196,7 +197,7 @@ protected Object filterValue(Object value, String dataType) throws SQLException
 
     public void addAndCheckCommit(TapRecordEvent recordEvent, WriteListResult<TapRecordEvent> listResult) {
         batchCacheSize++;
-        if (updatePolicy == LOG_ON_NONEXISTS && recordEvent instanceof TapUpdateRecordEvent) {
+        if (updatePolicy == LOG_ON_NONEXISTS && recordEvent instanceof TapUpdateRecordEvent || deletePolicy == LOG_ON_NONEXISTS && recordEvent instanceof TapDeleteRecordEvent) {
             batchCache.add(recordEvent);
         }
     }

connectors-common/sql-core/src/main/java/io/tapdata/common/dml/NormalRecordWriter.java

@@ -32,6 +32,7 @@ public class NormalRecordWriter {
     protected NormalWriteRecorder updateRecorder;
     protected String updatePolicy = ConnectionOptions.DML_UPDATE_POLICY_IGNORE_ON_NON_EXISTS;
     protected NormalWriteRecorder deleteRecorder;
+    protected String deletePolicy = ConnectionOptions.DML_DELETE_POLICY_IGNORE_ON_NON_EXISTS;
     protected String version;
     protected Connection connection;
     protected final TapTable tapTable;
@@ -79,6 +80,7 @@ public void write(List<TapRecordEvent> tapRecordEvents, Consumer<WriteListResult
             updateRecorder.setUpdatePolicy(updatePolicy);
             updateRecorder.setTapLogger(tapLogger);
             deleteRecorder.setVersion(version);
+            deleteRecorder.setDeletePolicy(deletePolicy);
             deleteRecorder.setTapLogger(tapLogger);
             //doubleActive
             if (Boolean.TRUE.equals(commonDbConfig.getDoubleActive())) {
@@ -185,6 +187,11 @@ public NormalRecordWriter setUpdatePolicy(String updatePolicy) {
         return this;
     }
 
+    public NormalRecordWriter setDeletePolicy(String deletePolicy) {
+        this.deletePolicy = deletePolicy;
+        return this;
+    }
+
     public NormalRecordWriter setTapLogger(Log tapLogger) {
         this.tapLogger = tapLogger;
         return this;

connectors-common/sql-core/src/main/java/io/tapdata/common/dml/NormalWriteRecorder.java

@@ -1,6 +1,7 @@
 package io.tapdata.common.dml;
 
 import io.netty.buffer.ByteBuf;
+import io.tapdata.entity.event.dml.TapDeleteRecordEvent;
 import io.tapdata.entity.event.dml.TapRecordEvent;
 import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
 import io.tapdata.entity.logger.Log;
@@ -45,6 +46,7 @@ public abstract class NormalWriteRecorder {
     protected Boolean fileInput = false;
     protected ByteBuf buffer;
     protected WritePolicyEnum updatePolicy;
+    protected WritePolicyEnum deletePolicy;
     protected char escapeChar = '"';
 
     protected String preparedStatementKey;
@@ -136,7 +138,17 @@ public void executeBatch(WriteListResult<TapRecordEvent> listResult) throws SQLE
                     while (iterator.hasNext()) {
                         TapRecordEvent event = iterator.next();
                         if (0 >= writeResults[index++]) {
-                            tapLogger.info("update record ignored: {}", event);
+                            tapLogger.warn("update record ignored: {}", event);
+                        }
+                    }
+                }
+                if (LOG_ON_NONEXISTS == deletePolicy) {
+                    Iterator<TapRecordEvent> iterator = batchCache.iterator();
+                    int index = 0;
+                    while (iterator.hasNext()) {
+                        TapRecordEvent event = iterator.next();
+                        if (0 >= writeResults[index++]) {
+                            tapLogger.warn("delete record ignored: {}", event);
                         }
                     }
                 }
@@ -155,7 +167,7 @@ public void executeBatch(WriteListResult<TapRecordEvent> listResult) throws SQLE
     //commit when cacheSize >= 1000
     public void addAndCheckCommit(TapRecordEvent recordEvent, WriteListResult<TapRecordEvent> listResult) throws SQLException {
         batchCacheSize++;
-        if (updatePolicy == LOG_ON_NONEXISTS && recordEvent instanceof TapUpdateRecordEvent) {
+        if (updatePolicy == LOG_ON_NONEXISTS && recordEvent instanceof TapUpdateRecordEvent || deletePolicy == LOG_ON_NONEXISTS && recordEvent instanceof TapDeleteRecordEvent) {
             batchCache.add(recordEvent);
         }
         if (batchCacheSize >= 1000) {
@@ -189,6 +201,10 @@ public void setUpdatePolicy(String updatePolicy) {
         this.updatePolicy = WritePolicyEnum.valueOf(updatePolicy.toUpperCase());
     }
 
+    public void setDeletePolicy(String deletePolicy) {
+        this.deletePolicy = WritePolicyEnum.valueOf(deletePolicy.toUpperCase());
+    }
+
     public void setTapLogger(Log tapLogger) {
         this.tapLogger = tapLogger;
     }

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

@@ -444,6 +444,10 @@ protected void writeRecord(TapConnectorContext connectorContext, List<TapRecordE
         if (updateDmlPolicy == null) {
             updateDmlPolicy = ConnectionOptions.DML_UPDATE_POLICY_IGNORE_ON_NON_EXISTS;
         }
+        String deleteDmlPolicy = connectorContext.getConnectorCapabilities().getCapabilityAlternative(ConnectionOptions.DML_DELETE_POLICY);
+        if (deleteDmlPolicy == null) {
+            deleteDmlPolicy = ConnectionOptions.DML_DELETE_POLICY_IGNORE_ON_NON_EXISTS;
+        }
         NormalRecordWriter postgresRecordWriter;
         if (isTransaction) {
             String threadName = Thread.currentThread().getName();
@@ -457,11 +461,13 @@ protected void writeRecord(TapConnectorContext connectorContext, List<TapRecordE
             postgresRecordWriter = new PostgresRecordWriter(postgresJdbcContext, connection, tapTable, hasUniqueIndex ? postgresVersion : "90500")
                     .setInsertPolicy(insertDmlPolicy)
                     .setUpdatePolicy(updateDmlPolicy)
+                    .setDeletePolicy(deleteDmlPolicy)
                     .setTapLogger(tapLogger);
         } else {
             postgresRecordWriter = new PostgresRecordWriter(postgresJdbcContext, tapTable, hasUniqueIndex ? postgresVersion : "90500")
                     .setInsertPolicy(insertDmlPolicy)
                     .setUpdatePolicy(updateDmlPolicy)
+                    .setDeletePolicy(deleteDmlPolicy)
                     .setTapLogger(tapLogger);
         }
         postgresRecordWriter.closeConstraintCheck();

connectors/postgres-connector/src/main/resources/spec_postgres.json

@@ -16,6 +16,10 @@
         "id": "dml_update_policy",
         "alternatives": ["ignore_on_nonexists", "insert_on_nonexists", "log_on_nonexists"]
       },
+      {
+        "id": "dml_delete_policy",
+        "alternatives": ["ignore_on_nonexists", "log_on_nonexists"]
+      },
       {
         "id": "dml_check_policy",
         "alternatives": ["ignore_all_check", "default_check"]