[flink] Fix that action/procedure cannot remove unexisting files from manifests when dv enabled. (#6854)

Author: zhoulii

paimon-core/src/main/java/org/apache/paimon/operation/ListUnexistingFiles.java

@@ -27,6 +27,7 @@
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.utils.FileOperationThreadPool;
 import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.Filter;
 import org.apache.paimon.utils.ThreadPoolUtils;
 
 import java.io.IOException;
@@ -58,6 +59,7 @@ public Map<Integer, Map<String, DataFileMeta>> list(BinaryRow partition) throws
         Map<Integer, Map<String, DataFileMeta>> result = new HashMap<>();
         List<Split> splits =
                 table.newScan()
+                        .withLevelFilter(Filter.alwaysTrue())
                         .withPartitionFilter(Collections.singletonList(partition))
                         .plan()
                         .splits();

paimon-core/src/test/java/org/apache/paimon/operation/ListUnexistingFilesTest.java

@@ -38,7 +38,8 @@
 
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -50,23 +51,30 @@
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
 
 /** Tests for {@link ListUnexistingFiles}. */
 public class ListUnexistingFilesTest {
 
     @TempDir java.nio.file.Path tempDir;
 
+    private static Stream<Arguments> params() {
+        return Stream.of(
+                arguments(-1, false), arguments(3, false), arguments(-1, true), arguments(3, true));
+    }
+
     @ParameterizedTest
-    @ValueSource(ints = {-1, 3})
-    public void testListFiles(int bucket) throws Exception {
+    @MethodSource("params")
+    public void testListFiles(int bucket, boolean dvEnabled) throws Exception {
         int numPartitions = 2;
         int numFiles = 10;
         int[] numDeletes = new int[numPartitions];
         FileStoreTable table =
                 prepareRandomlyDeletedTable(
-                        tempDir.toString(), "mydb", "t", bucket, numFiles, numDeletes);
+                        tempDir.toString(), "mydb", "t", bucket, numFiles, numDeletes, dvEnabled);
 
         Function<Integer, BinaryRow> binaryRow =
                 i -> {
@@ -90,7 +98,8 @@ public static FileStoreTable prepareRandomlyDeletedTable(
             String tableName,
             int bucket,
             int numFiles,
-            int[] numDeletes)
+            int[] numDeletes,
+            boolean dvEnabled)
             throws Exception {
         RowType rowType =
                 RowType.of(
@@ -102,12 +111,16 @@ public static FileStoreTable prepareRandomlyDeletedTable(
         if (bucket > 0) {
             options.put(CoreOptions.BUCKET_KEY.key(), "id");
         }
+        if (dvEnabled) {
+            options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+        }
         FileStoreTable table =
                 createPaimonTable(
                         warehouse,
                         databaseName,
                         tableName,
                         rowType,
+                        dvEnabled ? Arrays.asList("pt", "id") : Collections.emptyList(),
                         Collections.singletonList("pt"),
                         options);
 
@@ -124,7 +137,11 @@ public static FileStoreTable prepareRandomlyDeletedTable(
         int identifier = 0;
         for (int i = 0; i < numPartitions; i++) {
             for (int j = 0; j < numFiles; j++) {
-                write.write(GenericRow.of(i, random.nextInt(), random.nextLong()));
+                if (dvEnabled && bucket == -1) {
+                    write.write(GenericRow.of(i, random.nextInt(), random.nextLong()), 0);
+                } else {
+                    write.write(GenericRow.of(i, random.nextInt(), random.nextLong()));
+                }
                 identifier++;
                 commit.commit(identifier, write.prepareCommit(false, identifier));
             }
@@ -157,19 +174,15 @@ private static FileStoreTable createPaimonTable(
             String databaseName,
             String tableName,
             RowType rowType,
+            List<String> primaryKeys,
             List<String> partitionKeys,
             Map<String, String> customOptions)
             throws Exception {
         LocalFileIO fileIO = LocalFileIO.create();
         Path path = new Path(warehouse);
 
         Schema schema =
-                new Schema(
-                        rowType.getFields(),
-                        partitionKeys,
-                        Collections.emptyList(),
-                        customOptions,
-                        "");
+                new Schema(rowType.getFields(), partitionKeys, primaryKeys, customOptions, "");
 
         try (FileSystemCatalog paimonCatalog = new FileSystemCatalog(fileIO, path)) {
             paimonCatalog.createDatabase(databaseName, true);

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveUnexistingFilesActionITCase.java

@@ -41,7 +41,7 @@ public void testAction(int bucket) throws Exception {
         int numFiles = 10;
         int[] numDeletes = new int[numPartitions];
         ListUnexistingFilesTest.prepareRandomlyDeletedTable(
-                warehouse, "mydb", "t", bucket, numFiles, numDeletes);
+                warehouse, "mydb", "t", bucket, numFiles, numDeletes, false);
 
         RemoveUnexistingFilesAction action =
                 createAction(

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RemoveUnexistingFilesProcedureITCase.java

@@ -43,7 +43,7 @@ public void testProcedure(int bucket) throws Exception {
         int numFiles = 10;
         int[] numDeletes = new int[numPartitions];
         ListUnexistingFilesTest.prepareRandomlyDeletedTable(
-                warehouse, "mydb", "t", bucket, numFiles, numDeletes);
+                warehouse, "mydb", "t", bucket, numFiles, numDeletes, false);
 
         TableEnvironment tEnv = tableEnvironmentBuilder().batchMode().build();
         tEnv.executeSql(

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveUnexistingFilesProcedureTest.scala

@@ -46,7 +46,8 @@ class RemoveUnexistingFilesProcedureTest extends PaimonSparkTestBase {
       tableName,
       bucket,
       numFiles,
-      numDeletes)
+      numDeletes,
+      false)
 
     val actual = new Array[Int](numPartitions)
     val pattern = "pt=(\\d+?)/".r