From 096bac84f8aeff1c2fbdb74bbff829441795e631 Mon Sep 17 00:00:00 2001
From: zenghua <huazeng@dmetasoul.com>
Date: Wed, 31 Jul 2024 15:12:36 +0800
Subject: [PATCH] merge LakeSoulMultiTableSinkGlobalCommittable with sorted
 createtime

Signed-off-by: zenghua <huazeng@dmetasoul.com>
---
 ...akeSoulMultiTableSinkGlobalCommittable.java | 18 ++++++++++++------
 1 file changed, 12 insertions(+), 6 deletions(-)

diff --git a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/state/LakeSoulMultiTableSinkGlobalCommittable.java b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/state/LakeSoulMultiTableSinkGlobalCommittable.java
index f20e0c251..a3c400230 100644
--- a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/state/LakeSoulMultiTableSinkGlobalCommittable.java
+++ b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/state/LakeSoulMultiTableSinkGlobalCommittable.java
@@ -49,18 +49,24 @@ public LakeSoulMultiTableSinkGlobalCommittable(
     }
 
     public static LakeSoulMultiTableSinkGlobalCommittable fromLakeSoulMultiTableSinkGlobalCommittable(
-            List<LakeSoulMultiTableSinkGlobalCommittable> globalCommittables, boolean isBounded) {
-        Map<Tuple2<TableSchemaIdentity, String>, List<LakeSoulMultiTableSinkCommittable>> groupedCommitables =
-                new HashMap<>();
-        globalCommittables.forEach(globalCommittable -> globalCommittable.getGroupedCommittable().forEach(
-                (key, value) -> groupedCommitables.computeIfAbsent(key, tuple2 -> new ArrayList<>()).addAll(value)));
-        return new LakeSoulMultiTableSinkGlobalCommittable(groupedCommitables, isBounded);
+            List<LakeSoulMultiTableSinkGlobalCommittable> globalCommittableList, boolean isBounded) {
+        List<LakeSoulMultiTableSinkCommittable> committableList = new ArrayList<>();
+        for (LakeSoulMultiTableSinkGlobalCommittable globalCommittable : globalCommittableList) {
+            globalCommittable.getGroupedCommittable().values().forEach(committableList::addAll);
+        }
+        return fromLakeSoulMultiTableSinkCommittable(committableList, isBounded);
+//        Map<Tuple2<TableSchemaIdentity, String>, List<LakeSoulMultiTableSinkCommittable>> groupedCommitables =
+//                new HashMap<>();
+//        globalCommittables.forEach(globalCommittable -> globalCommittable.getGroupedCommittable().forEach(
+//                (key, value) -> groupedCommitables.computeIfAbsent(key, tuple2 -> new ArrayList<>()).addAll(value)));
+//        return new LakeSoulMultiTableSinkGlobalCommittable(groupedCommitables, isBounded);
     }
 
     public static LakeSoulMultiTableSinkGlobalCommittable fromLakeSoulMultiTableSinkCommittable(
             List<LakeSoulMultiTableSinkCommittable> committables, boolean isBounded) {
         Map<Tuple2<TableSchemaIdentity, String>, List<LakeSoulMultiTableSinkCommittable>> groupedCommitables =
                 new HashMap<>();
+        committables.sort(LakeSoulMultiTableSinkCommittable::compareTo);
         committables.forEach(committable -> groupedCommitables.computeIfAbsent(
                         Tuple2.of(committable.getIdentity(), committable.getBucketId()), tuple2 -> new ArrayList<>())
                 .add(committable));