[fix] Fix IndexFileHandler to support deleting all index types during snapshot expiration

paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java

@@ -33,6 +33,7 @@
 import org.apache.paimon.utils.SnapshotManager;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -41,7 +42,6 @@
 import java.util.Optional;
 import java.util.Set;
 
-import static org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
 import static org.apache.paimon.index.HashIndexFile.HASH_INDEX;
 
 /** Handle index files. */
@@ -205,24 +205,16 @@ public List<IndexManifestEntry> readManifestWithIOException(String indexManifest
         return indexManifestFile.readWithIOException(indexManifest);
     }
 
-    private IndexFile indexFile(IndexManifestEntry entry) {
-        IndexFileMeta file = entry.indexFile();
-        switch (file.indexType()) {
-            case HASH_INDEX:
-                return hashIndex(entry.partition(), entry.bucket());
-            case DELETION_VECTORS_INDEX:
-                return dvIndex(entry.partition(), entry.bucket());
-            default:
-                throw new IllegalArgumentException("Unknown index type: " + file.indexType());
+    public boolean existsIndexFile(IndexManifestEntry entry) {
+        try {
+            return fileIO.exists(filePath(entry));
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
         }
     }
 
-    public boolean existsIndexFile(IndexManifestEntry file) {
-        return indexFile(file).exists(file.indexFile());
-    }
-
     public void deleteIndexFile(IndexManifestEntry entry) {
-        indexFile(entry).delete(entry.indexFile());
+        fileIO.deleteQuietly(filePath(entry));
     }
 
     public void deleteManifest(String indexManifest) {

paimon-core/src/test/java/org/apache/paimon/index/IndexFileHandlerTest.java

@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.index;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.IndexFilePathFactories;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Test for {@link IndexFileHandler#deleteIndexFile} and {@link IndexFileHandler#existsIndexFile}.
+ */
+public class IndexFileHandlerTest {
+
+    @TempDir java.nio.file.Path tempPath;
+
+    private FileIO fileIO;
+    private IndexFileHandler handler;
+
+    @BeforeEach
+    void setUp() {
+        fileIO = LocalFileIO.create();
+        Path root = new Path(tempPath.toUri());
+
+        FileStorePathFactory pathFactory =
+                new FileStorePathFactory(
+                        root,
+                        RowType.builder().build(),
+                        "default",
+                        CoreOptions.FILE_FORMAT.defaultValue().toString(),
+                        CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+                        CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(),
+                        CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(),
+                        CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(),
+                        CoreOptions.FILE_COMPRESSION.defaultValue(),
+                        null,
+                        null,
+                        CoreOptions.ExternalPathStrategy.NONE,
+                        null,
+                        false,
+                        null);
+
+        handler =
+                new IndexFileHandler(
+                        fileIO,
+                        null,
+                        null,
+                        new IndexFilePathFactories(pathFactory),
+                        MemorySize.ofMebiBytes(2),
+                        false);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"HASH", "DELETION_VECTORS", "btree", "bitmap", "lumina-vector-ann"})
+    void testExistsAndDeleteIndexFile(String indexType) throws IOException {
+        String fileName = "index-" + UUID.randomUUID();
+        IndexFileMeta meta =
+                new IndexFileMeta(indexType, fileName, 4, 1, (GlobalIndexMeta) null, null);
+        IndexManifestEntry entry =
+                new IndexManifestEntry(FileKind.ADD, BinaryRow.EMPTY_ROW, 0, meta);
+
+        // create the file at the expected path
+        Path filePath = handler.filePath(entry);
+        fileIO.tryToWriteAtomic(filePath, "test");
+
+        assertThat(handler.existsIndexFile(entry)).isTrue();
+
+        handler.deleteIndexFile(entry);
+
+        assertThat(handler.existsIndexFile(entry)).isFalse();
+    }
+}