[core] Remote SST should work for upgrade files

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/ChangelogMergeTreeRewriter.java

@@ -184,9 +184,10 @@ private CompactResult rewriteOrProduceChangelog(
 
         if (rewriteCompactFile) {
             notifyRewriteCompactBefore(before);
-            after = notifyRewriteCompactAfter(after);
         }
 
+        after = notifyRewriteCompactAfter(after);
+
         List<DataFileMeta> changelogFiles =
                 changelogFileWriter != null
                         ? changelogFileWriter.result()

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupRemoteFileTableTest.java

@@ -120,21 +120,13 @@ private void innerTestRemoteFile(boolean schemaEvolution, boolean notCompatible)
         ReadBuilder readBuilder = table.newReadBuilder();
         List<Split> splits = readBuilder.newScan().plan().splits();
         assertThat(splits).hasSize(1);
-        DataSplit firstSplit = (DataSplit) splits.get(0);
-        DataFileMeta firstFile = firstSplit.dataFiles().get(0);
-        List<String> extraFiles = firstFile.extraFiles();
-        String extraFile = extraFiles.get(0);
-        // data-410685c7-4cc2-47d7-9dec-393f6cfe9d64-0.parquet.115.position.v1.lookup
-        assertThat(extraFile).endsWith(".position.v1.lookup");
-        long lookupFileSize =
-                fileIO.getFileSize(
-                        new Path(new Path(tempPath.toUri()), "default.db/t/bucket-0/" + extraFile));
-        String[] split = extraFile.split("\\.");
-        assertThat(split[split.length - 4]).isEqualTo(String.valueOf(lookupFileSize));
+        allShouldHaveRemoteSst(splits);
 
         // third write with lookup but no data file
 
         // delete file first
+        DataSplit firstSplit = (DataSplit) splits.get(0);
+        DataFileMeta firstFile = firstSplit.dataFiles().get(0);
         Path firstPath =
                 table.store()
                         .pathFactory()
@@ -174,9 +166,12 @@ private void innerTestRemoteFile(boolean schemaEvolution, boolean notCompatible)
             }
         }
 
+        // check remote lookup files
+        splits = readBuilder.newScan().plan().splits();
+        allShouldHaveRemoteSst(splits);
+
         // restore file and check reading
         fileIO.copyFile(tmpPath, firstPath, false);
-        splits = readBuilder.newScan().plan().splits();
         List<GenericRow> result = new ArrayList<>();
         readBuilder
                 .newRead()
@@ -191,6 +186,31 @@ private void innerTestRemoteFile(boolean schemaEvolution, boolean notCompatible)
                         GenericRow.of(5, 1));
     }
 
+    private void allShouldHaveRemoteSst(List<Split> check) {
+        for (Split split : check) {
+            for (DataFileMeta file : ((DataSplit) split).dataFiles()) {
+                List<String> extraFiles = file.extraFiles();
+                String extraFile = extraFiles.get(0);
+                // data-410685c7-4cc2-47d7-9dec-393f6cfe9d64-0.parquet.115.position.v1.lookup
+                assertThat(extraFile).endsWith(".position.v1.lookup");
+                long lookupFileSize;
+                try {
+                    lookupFileSize =
+                            LocalFileIO.create()
+                                    .getFileSize(
+                                            new Path(
+                                                    new Path(tempPath.toUri()),
+                                                    "default.db/t/bucket-0/" + extraFile));
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
+                }
+                String[] extraFileSplit = extraFile.split("\\.");
+                assertThat(extraFileSplit[extraFileSplit.length - 4])
+                        .isEqualTo(String.valueOf(lookupFileSize));
+            }
+        }
+    }
+
     @Test
     public void testRemoteFileLevelThreshold() throws Exception {
         Options options = new Options();