[fix](ccr)fix ccr canot handle insert default list partition

fe/fe-core/src/main/java/org/apache/doris/binlog/UpsertRecord.java

@@ -40,6 +40,9 @@ public static class PartitionRecord {
             @SerializedName(value = "range")
             private String range;
 
+            @SerializedName(value = "rangeNotDefault")
+            private String rangeNotDefault;
+
             @SerializedName(value = "version")
             public long version;
 
@@ -74,6 +77,7 @@ private void addPartitionRecord(long subTxnId, PartitionCommitInfo partitionComm
             partitionRecord.subTxnId = subTxnId;
             partitionRecord.partitionId = partitionCommitInfo.getPartitionId();
             partitionRecord.range = partitionCommitInfo.getPartitionRange();
+            partitionRecord.rangeNotDefault = partitionCommitInfo.getNotDefaultRange();
             partitionRecord.version = partitionCommitInfo.getVersion();
             partitionRecord.isTemp = partitionCommitInfo.isTempPartition();
             partitionRecords.add(partitionRecord);

fe/fe-core/src/main/java/org/apache/doris/catalog/ListPartitionItem.java

@@ -69,6 +69,19 @@ public String getItemsString() {
         return getItems().toString();
     }
 
+    public String getItemsStringSkipDefaultKeyCheck() {
+        // ATTN: DO NOT EDIT unless unless you explicitly guarantee compatibility
+        // between different versions.
+        //
+        // the ccr syncer depends on this string to identify partitions between two
+        // clusters (cluster versions may be different).
+        List<String> list = new ArrayList<>();
+        for (PartitionKey partitionKey : partitionKeys) {
+            list.add(partitionKey.toStringSkipDefaultKeyCheck());
+        }
+        return list.toString();
+    }
+
     public String getItemsSql() {
         return toSql();
     }

fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionInfo.java

@@ -178,6 +178,16 @@ public String getPartitionRangeString(long partitionId) {
         return partitionRange;
     }
 
+    // Get the unique string of the partition range skip default check.
+    public String getPartitionRangeSkipDefaultKeyString(long partitionId) {
+        String partitionRange = "";
+        if (getType() == PartitionType.RANGE || getType() == PartitionType.LIST) {
+            PartitionItem item = getItem(partitionId);
+            partitionRange = item.getItemsStringSkipDefaultKeyCheck();
+        }
+        return partitionRange;
+    }
+
     public PartitionItem getItemOrAnalysisException(long partitionId) throws AnalysisException {
         PartitionItem item = idToItem.get(partitionId);
         if (item == null) {

fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionItem.java

@@ -31,6 +31,8 @@ public abstract class PartitionItem implements Comparable<PartitionItem> {
     // get the unique string of the partition item.
     public abstract String getItemsString();
 
+    public abstract String getItemsStringSkipDefaultKeyCheck();
+
     public abstract <T> T getItems();
 
     public abstract PartitionItem getIntersect(PartitionItem newItem);

fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionKey.java

@@ -414,6 +414,24 @@ public String toString() {
         return builder.toString();
     }
 
+    public String toStringSkipDefaultKeyCheck() {
+        // ATTN: DO NOT EDIT unless unless you explicitly guarantee compatibility
+        // between different versions.
+        //
+        // the ccr syncer depends on this string to identify partitions between two
+        // clusters (cluster versions may be different).
+        StringBuilder builder = new StringBuilder();
+        builder.append("types: [");
+        builder.append(Joiner.on(", ").join(types));
+        builder.append("]; ");
+
+        builder.append("keys: [");
+        builder.append(toString(keys));
+        builder.append("]; ");
+
+        return builder.toString();
+    }
+
     public static String toString(List<LiteralExpr> keys) {
         StringBuilder builder = new StringBuilder();
         int i = 0;

fe/fe-core/src/main/java/org/apache/doris/catalog/RangePartitionItem.java

@@ -54,6 +54,15 @@ public String getItemsString() {
         return partitionKeyRange.toString();
     }
 
+    public String getItemsStringSkipDefaultKeyCheck() {
+        // ATTN: DO NOT EDIT unless unless you explicitly guarantee compatibility
+        // between different versions.
+        //
+        // the ccr syncer depends on this string to identify partitions between two
+        // clusters (cluster versions may be different).
+        return getItemsString();
+    }
+
     public String getItemsSql() {
         return toPartitionKeyDesc().toSql();
     }

fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java

@@ -1485,8 +1485,9 @@ protected void unprotectedPreCommitTransaction2PC(TransactionState transactionSt
             PartitionInfo tblPartitionInfo = table.getPartitionInfo();
             for (long partitionId : tableToPartition.get(tableId)) {
                 String partitionRange = tblPartitionInfo.getPartitionRangeString(partitionId);
+                String notDefaultKeyRange = tblPartitionInfo.getPartitionRangeSkipDefaultKeyString(partitionId);
                 PartitionCommitInfo partitionCommitInfo = new PartitionCommitInfo(
-                        partitionId, partitionRange, -1, -1,
+                        partitionId, partitionRange, notDefaultKeyRange, -1, -1,
                         table.isTemporaryPartition(partitionId));
                 tableCommitInfo.addPartitionCommitInfo(partitionCommitInfo);
             }
@@ -1500,7 +1501,8 @@ protected void unprotectedPreCommitTransaction2PC(TransactionState transactionSt
     private PartitionCommitInfo generatePartitionCommitInfo(OlapTable table, long partitionId, long partitionVersion) {
         PartitionInfo tblPartitionInfo = table.getPartitionInfo();
         String partitionRange = tblPartitionInfo.getPartitionRangeString(partitionId);
-        return new PartitionCommitInfo(partitionId, partitionRange,
+        String notDefaultKeyRange = tblPartitionInfo.getPartitionRangeSkipDefaultKeyString(partitionId);
+        return new PartitionCommitInfo(partitionId, partitionRange, notDefaultKeyRange,
                 partitionVersion, System.currentTimeMillis() /* use as partition visible time */,
                 table.isTemporaryPartition(partitionId));
     }

fe/fe-core/src/main/java/org/apache/doris/transaction/PartitionCommitInfo.java

@@ -33,6 +33,8 @@ public class PartitionCommitInfo implements Writable {
     private long partitionId;
     @SerializedName(value = "range")
     private String range;
+    @SerializedName(value = "rangeNotDefault")
+    private String rangeNotDefault;
     @SerializedName(value = "version")
     private long version;
     @SerializedName(value = "versionTime")
@@ -44,11 +46,12 @@ public PartitionCommitInfo() {
 
     }
 
-    public PartitionCommitInfo(long partitionId, String partitionRange, long version, long visibleTime,
-            boolean isTempPartition) {
+    public PartitionCommitInfo(long partitionId, String partitionRange, String notDefaultRange, long version,
+            long visibleTime, boolean isTempPartition) {
         super();
         this.partitionId = partitionId;
         this.range = partitionRange;
+        this.rangeNotDefault = notDefaultRange;
         this.version = version;
         this.versionTime = visibleTime;
         this.isTempPartition = isTempPartition;
@@ -73,6 +76,10 @@ public String getPartitionRange() {
         return range;
     }
 
+    public String getNotDefaultRange() {
+        return rangeNotDefault;
+    }
+
     public long getVersion() {
         return version;
     }