[Flink] Fix clean service: updateCommit as old compactionCommit (lakesoul-io#669)

Signed-off-by: ChenYunHey <[email protected]>
Co-authored-by: Xu Chen <[email protected]>

Author: ChenYunHey

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/entry/clean/NewCleanJob.java

@@ -179,7 +179,7 @@ public void processElement(PartitionInfo value, KeyedProcessFunction<String, Par
             List<String> snapshot = value.snapshot;
             if (commitOp.equals("CompactionCommit") || commitOp.equals("UpdateCommit") ){
                 if ( snapshot.size() == 1) {
-                    boolean isOldCompaction = cleanUtils.getCompactVersion(tableId, partitionDesc, version, pgConnection);
+                    boolean isOldCompaction = commitOp.equals("UpdateCommit") || cleanUtils.getCompactVersion(tableId, partitionDesc, version, pgConnection);
                     log.info("当前识别出来为旧版压缩："+ isOldCompaction);
                     compactionVersionState.update(isOldCompaction);
                 }