refractor lakesoul sink state

Signed-off-by: zenghua <[email protected]>
Ceng23333 · Mar 14, 2024 · 308fead · 308fead
1 parent 50344cd
commit 308fead
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 5 deletions.
diff --git a/...l-flink/src/main/java/org/apache/flink/lakesoul/sink/state/LakeSoulWriterBucketState.java b/...l-flink/src/main/java/org/apache/flink/lakesoul/sink/state/LakeSoulWriterBucketState.java
@@ -46,6 +46,7 @@ public LakeSoulWriterBucketState(
 
     public LakeSoulWriterBucketState(
             TableSchemaIdentity identity,
+            Path bucketPath,
             HashMap<String, List<InProgressFileWriter.PendingFileRecoverable>> pendingFileRecoverableMap
     ) {
         this.identity = identity;
@@ -55,7 +56,7 @@ public LakeSoulWriterBucketState(
         } else {
             this.bucketId = "";
         }
-        this.bucketPath = null;
+        this.bucketPath = bucketPath;
 
         this.pendingFileRecoverableMap = pendingFileRecoverableMap;
     }

diff --git a/...c/main/java/org/apache/flink/lakesoul/sink/state/LakeSoulWriterBucketStateSerializer.java b/...c/main/java/org/apache/flink/lakesoul/sink/state/LakeSoulWriterBucketStateSerializer.java
@@ -67,7 +67,7 @@ public LakeSoulWriterBucketState deserialize(int version, byte[] serialized) thr
     private void serialize(LakeSoulWriterBucketState state, DataOutputView dataOutputView)
             throws IOException {
 //        dataOutputView.writeUTF(state.getBucketId());
-//        dataOutputView.writeUTF(state.getBucketPath().toString());
+        dataOutputView.writeUTF(state.getBucketPath().toString());
 
         SimpleVersionedSerialization.writeVersionAndSerialize(
                 tableSchemaIdentitySerializer, state.getIdentity(), dataOutputView);
@@ -102,7 +102,7 @@ private LakeSoulWriterBucketState internalDeserialize(
             throws IOException {
 
 //        String bucketId = dataInputView.readUTF();
-//        String bucketPathStr = dataInputView.readUTF();
+        String bucketPathStr = dataInputView.readUTF();
 
         TableSchemaIdentity identity = SimpleVersionedSerialization.readVersionAndDeSerialize(
                 tableSchemaIdentitySerializer, dataInputView);
@@ -119,9 +119,10 @@ private LakeSoulWriterBucketState internalDeserialize(
             }
             pendingFileRecoverableMap.put(bucketId, pendingFileRecoverableList);
         }
-        
+
         return new LakeSoulWriterBucketState(
                 identity,
+                new Path(bucketPathStr),
                 pendingFileRecoverableMap);
     }
 

diff --git a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/writer/LakeSoulWriterBucket.java b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/writer/LakeSoulWriterBucket.java
@@ -206,7 +206,7 @@ LakeSoulWriterBucketState snapshotState() throws IOException {
 //                getBucketId(),
 //                bucketPath,
 //                tmpPending);
-        return new LakeSoulWriterBucketState(tableId, new HashMap<>(pendingFilesMap));
+        return new LakeSoulWriterBucketState(tableId, bucketPath, new HashMap<>(pendingFilesMap));
     }
 
     void onProcessingTime(long timestamp) throws IOException {