[FLINK-38833][paimon] Shuffle record to different subtasks by table, partition and bucket. (apache#4298)

Signed-off-by: Pei Yu <125331682@qq.com>

Author: sd4324530

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonEventSink.java

@@ -69,6 +69,7 @@ public DataStream<Event> addPreWriteTopology(DataStream<Event> dataStream) {
                             if (event instanceof BucketWrapperChangeEvent) {
                                 // Add hash of tableId to avoid data skew.
                                 return ((BucketWrapperChangeEvent) event).getBucket()
+                                        + ((BucketWrapperChangeEvent) event).getPartition()
                                         + ((BucketWrapperChangeEvent) event).tableId().hashCode();
                             } else {
                                 return ((BucketWrapper) event).getBucket();

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java

@@ -157,6 +157,7 @@ public void processElement(StreamRecord<Event> streamRecord) throws Exception {
                     bucketAssignerMap.computeIfAbsent(
                             dataChangeEvent.tableId(), this::getTableInfo);
             int bucket;
+            int partition;
             GenericRow genericRow =
                     PaimonWriterHelper.convertEventToGenericRow(
                             dataChangeEvent,
@@ -171,17 +172,20 @@ public void processElement(StreamRecord<Event> streamRecord) throws Exception {
                                 tuple4.f2.assign(
                                         tuple4.f3.partition(genericRow),
                                         tuple4.f3.trimmedPrimaryKey(genericRow).hashCode());
+                        partition = tuple4.f3.partition(genericRow).hashCode();
                         break;
                     }
                 case HASH_FIXED:
                     {
                         tuple4.f1.setRecord(genericRow);
                         bucket = tuple4.f1.bucket();
+                        partition = tuple4.f1.partition().hashCode();
                         break;
                     }
                 case BUCKET_UNAWARE:
                     {
                         bucket = 0;
+                        partition = 0;
                         break;
                     }
                 case KEY_DYNAMIC:
@@ -191,14 +195,16 @@ public void processElement(StreamRecord<Event> streamRecord) throws Exception {
                     }
             }
             output.collect(
-                    new StreamRecord<>(new BucketWrapperChangeEvent(bucket, dataChangeEvent)));
+                    new StreamRecord<>(
+                            new BucketWrapperChangeEvent(bucket, partition, dataChangeEvent)));
         } else {
             // Broadcast SchemachangeEvent.
             for (int index = 0; index < totalTasksNumber; index++) {
                 output.collect(
                         new StreamRecord<>(
                                 new BucketWrapperChangeEvent(
                                         index,
+                                        0,
                                         convertSchemaChangeEvent((SchemaChangeEvent) event))));
             }
         }
@@ -279,6 +285,7 @@ public DataChangeEvent convertDataChangeEvent(DataChangeEvent dataChangeEvent)
                             new StreamRecord<>(
                                     new BucketWrapperChangeEvent(
                                             index,
+                                            0,
                                             new CreateTableEvent(
                                                     tableId,
                                                     mixedSchemaInfo.paimonSchemaInfo

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketWrapperChangeEvent.java

@@ -27,18 +27,24 @@
 public class BucketWrapperChangeEvent implements ChangeEvent, BucketWrapper, Serializable {
     private static final long serialVersionUID = 1L;
     private final int bucket;
+    private final int partition;
 
     private final ChangeEvent innerEvent;
 
-    public BucketWrapperChangeEvent(int bucket, ChangeEvent innerEvent) {
+    public BucketWrapperChangeEvent(int bucket, int partition, ChangeEvent innerEvent) {
         this.bucket = bucket;
+        this.partition = partition;
         this.innerEvent = innerEvent;
     }
 
     public int getBucket() {
         return bucket;
     }
 
+    public int getPartition() {
+        return partition;
+    }
+
     public ChangeEvent getInnerEvent() {
         return innerEvent;
     }
@@ -57,19 +63,23 @@ public boolean equals(Object o) {
             return false;
         }
         BucketWrapperChangeEvent that = (BucketWrapperChangeEvent) o;
-        return bucket == that.bucket && Objects.equals(innerEvent, that.innerEvent);
+        return bucket == that.bucket
+                && partition == that.partition
+                && Objects.equals(innerEvent, that.innerEvent);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(bucket, innerEvent);
+        return Objects.hash(bucket, partition, innerEvent);
     }
 
     @Override
     public String toString() {
         return "BucketWrapperChangeEvent{"
                 + "bucket="
                 + bucket
+                + ", partition="
+                + partition
                 + ", innerEvent="
                 + innerEvent
                 + '}';

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketWrapperEventSerializer.java

@@ -82,6 +82,7 @@ public void serialize(Event event, DataOutputView dataOutputView) throws IOExcep
             BucketWrapperChangeEvent bucketWrapperChangeEvent = (BucketWrapperChangeEvent) event;
             enumSerializer.serialize(EventClass.BUCKET_WRAPPER_CHANGE_EVENT, dataOutputView);
             dataOutputView.writeInt(bucketWrapperChangeEvent.getBucket());
+            dataOutputView.writeInt(bucketWrapperChangeEvent.getPartition());
             eventSerializer.serialize(bucketWrapperChangeEvent.getInnerEvent(), dataOutputView);
         } else if (event instanceof BucketWrapperFlushEvent) {
             enumSerializer.serialize(EventClass.BUCKET_WRAPPER_FLUSH_EVENT, dataOutputView);
@@ -107,7 +108,9 @@ public Event deserialize(DataInputView source) throws IOException {
                     schemaChangeEventTypeEnumSerializer.deserialize(source));
         } else {
             return new BucketWrapperChangeEvent(
-                    source.readInt(), (ChangeEvent) eventSerializer.deserialize(source));
+                    source.readInt(),
+                    source.readInt(),
+                    (ChangeEvent) eventSerializer.deserialize(source));
         }
     }