[Feature][Connectors-v2-Paimon] Adaptation Paimon 0.6 Version (#6061)

Author: SimonChou12138

seatunnel-connectors-v2/connector-paimon/pom.xml

@@ -30,7 +30,7 @@
     <name>SeaTunnel : Connectors V2 : Paimon</name>
 
     <properties>
-        <paimon.version>0.4.0-incubating</paimon.version>
+        <paimon.version>0.6.0-incubating</paimon.version>
     </properties>
 
     <dependencies>

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSink.java

@@ -26,6 +26,7 @@
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
 import org.apache.seatunnel.api.sink.SinkAggregatedCommitter;
 import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.common.config.CheckConfigUtil;
@@ -123,6 +124,11 @@ public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
         this.seaTunnelRowType = seaTunnelRowType;
     }
 
+    @Override
+    public SeaTunnelDataType<SeaTunnelRow> getConsumedType() {
+        return this.seaTunnelRowType;
+    }
+
     @Override
     public SinkWriter<SeaTunnelRow, PaimonCommitInfo, PaimonSinkState> createWriter(
             SinkWriter.Context context) throws IOException {

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkWriter.java

@@ -27,12 +27,11 @@
 import org.apache.seatunnel.connectors.seatunnel.paimon.utils.RowConverter;
 
 import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.operation.Lock;
 import org.apache.paimon.table.Table;
 import org.apache.paimon.table.sink.BatchTableCommit;
 import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
 import org.apache.paimon.table.sink.CommitMessage;
-import org.apache.paimon.table.sink.InnerTableCommit;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -51,6 +50,8 @@ public class PaimonSinkWriter
 
     private String commitUser = UUID.randomUUID().toString();
 
+    private final BatchWriteBuilder tableWriteBuilder;
+
     private final BatchTableWrite tableWrite;
 
     private long checkpointId = 0;
@@ -65,7 +66,8 @@ public class PaimonSinkWriter
 
     public PaimonSinkWriter(Context context, Table table, SeaTunnelRowType seaTunnelRowType) {
         this.table = table;
-        this.tableWrite = this.table.newBatchWriteBuilder().newWrite();
+        this.tableWriteBuilder = this.table.newBatchWriteBuilder().withOverwrite();
+        this.tableWrite = tableWriteBuilder.newWrite();
         this.seaTunnelRowType = seaTunnelRowType;
         this.context = context;
     }
@@ -76,17 +78,16 @@ public PaimonSinkWriter(
             SeaTunnelRowType seaTunnelRowType,
             List<PaimonSinkState> states) {
         this.table = table;
-        this.tableWrite = this.table.newBatchWriteBuilder().newWrite();
+        this.tableWriteBuilder = this.table.newBatchWriteBuilder().withOverwrite();
+        this.tableWrite = tableWriteBuilder.newWrite();
         this.seaTunnelRowType = seaTunnelRowType;
         this.context = context;
         if (Objects.isNull(states) || states.isEmpty()) {
             return;
         }
         this.commitUser = states.get(0).getCommitUser();
         this.checkpointId = states.get(0).getCheckpointId();
-        try (BatchTableCommit tableCommit =
-                ((InnerTableCommit) table.newBatchWriteBuilder().newCommit())
-                        .withLock(Lock.emptyFactory().create())) {
+        try (BatchTableCommit tableCommit = tableWriteBuilder.newCommit()) {
             List<CommitMessage> commitables =
                     states.stream()
                             .map(PaimonSinkState::getCommittables)

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/commit/PaimonAggregatedCommitter.java

@@ -25,7 +25,6 @@
 import org.apache.paimon.table.Table;
 import org.apache.paimon.table.sink.BatchTableCommit;
 import org.apache.paimon.table.sink.CommitMessage;
-import org.apache.paimon.table.sink.InnerTableCommit;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -54,8 +53,7 @@ public PaimonAggregatedCommitter(Table table) {
     public List<PaimonAggregatedCommitInfo> commit(
             List<PaimonAggregatedCommitInfo> aggregatedCommitInfo) throws IOException {
         try (BatchTableCommit tableCommit =
-                ((InnerTableCommit) table.newBatchWriteBuilder().newCommit())
-                        .withLock(localFactory.create())) {
+                table.newBatchWriteBuilder().withOverwrite().newCommit()) {
             List<CommitMessage> fileCommittables =
                     aggregatedCommitInfo.stream()
                             .map(PaimonAggregatedCommitInfo::getCommittables)