diff --git a/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java b/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java
index 84e48c553869..7c17ce1f57e0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java
@@ -118,11 +118,11 @@ public DataFileMeta process(BinaryRow partition, int bucket, ManifestEntry manif
                 maintainers.remove(name);
             } else {
                 Map<String, byte[]> indexTypeBytes = maintainers.get(name);
-                for (String indexType : entry.getValue().keySet()) {
-                    if (!indexTypeBytes.containsKey(indexType)) {
-                        indexTypeBytes.remove(indexType);
-                    }
-                }
+                Set<String> configuredIndexTypes =
+                        schemaInfo.projectedIndexTypes.getOrDefault(name, Collections.emptySet());
+                indexTypeBytes
+                        .keySet()
+                        .removeIf(indexType -> !configuredIndexTypes.contains(indexType));
             }
         }
 
@@ -180,7 +180,7 @@ public DataFileMeta process(BinaryRow partition, int bucket, ManifestEntry manif
         } else if (baos.size() == 0) {
             return dataFileMeta.copy(extras);
         } else {
-            return dataFileMeta.copy(baos.toByteArray());
+            return dataFileMeta.copy(extras).copy(baos.toByteArray());
         }
     }
 
@@ -214,6 +214,7 @@ public SchemaInfo schemaInfo(long schemaId) {
 
                 List<String> projectedColNames = new ArrayList<>();
                 Set<String> projectedColFullNames = new HashSet<>();
+                Map<String, Set<String>> projectedIndexTypes = new HashMap<>();
                 for (Map.Entry<FileIndexOptions.Column, Map<String, Options>> entry :
                         fileIndexOptions.entrySet()) {
                     FileIndexOptions.Column column = entry.getKey();
@@ -234,6 +235,9 @@ public SchemaInfo schemaInfo(long schemaId) {
                                             columnName, column.getNestedColumnName())
                                     : column.getColumnName();
                     projectedColFullNames.add(fullColumnName);
+                    projectedIndexTypes
+                            .computeIfAbsent(fullColumnName, ignored -> new HashSet<>())
+                            .addAll(entry.getValue().keySet());
                 }
 
                 schemaInfos.put(
@@ -244,7 +248,8 @@ public SchemaInfo schemaInfo(long schemaId) {
                                 projectedColNames.stream()
                                         .mapToInt(fileSchema::getFieldIndex)
                                         .toArray(),
-                                projectedColFullNames));
+                                projectedColFullNames,
+                                projectedIndexTypes));
                 fileSchemaIds.add(schemaId);
             }
 
@@ -276,16 +281,19 @@ private static class SchemaInfo {
         private final Map<String, String> colNameMapping;
         private final int[] projectedIndexCols;
         private final Set<String> projectedColFullNames;
+        private final Map<String, Set<String>> projectedIndexTypes;
 
         private SchemaInfo(
                 RowType fileSchema,
                 Map<String, String> colNameMapping,
                 int[] projectedIndexCols,
-                Set<String> projectedColFullNames) {
+                Set<String> projectedColFullNames,
+                Map<String, Set<String>> projectedIndexTypes) {
             this.fileSchema = fileSchema;
             this.colNameMapping = colNameMapping;
             this.projectedIndexCols = projectedIndexCols;
             this.projectedColFullNames = projectedColFullNames;
+            this.projectedIndexTypes = projectedIndexTypes;
         }
     }
 }
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedureITCase.java
index fe62eb3c6121..3f23c9078857 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedureITCase.java
@@ -18,11 +18,13 @@
 
 package org.apache.paimon.flink.procedure;
 
+import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.fileindex.FileIndexFormat;
 import org.apache.paimon.fileindex.FileIndexReader;
 import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.fs.ByteArraySeekableStream;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.io.DataFilePathFactory;
 import org.apache.paimon.manifest.ManifestEntry;
@@ -37,6 +39,7 @@
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
@@ -156,6 +159,78 @@ public void testPartitionFilter(boolean isNamedArgument) throws Exception {
         Assertions.assertThat(count.get()).isEqualTo(2);
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testFileIndexProcedureSwitchIndexType(boolean isNamedArgument) throws Exception {
+        sql(
+                "CREATE TABLE T ("
+                        + " k INT,"
+                        + " v STRING,"
+                        + " dt STRING"
+                        + ") PARTITIONED BY (dt) WITH ("
+                        + " 'write-only' = 'true',"
+                        + " 'file-index.bloom-filter.columns' = 'k',"
+                        + " 'bucket' = '-1'"
+                        + ")");
+        sql("INSERT INTO T VALUES (1, '100', '20221208')");
+
+        tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+        if (isNamedArgument) {
+            sql("CALL sys.rewrite_file_index(`table` => 'default.T')");
+        } else {
+            sql("CALL sys.rewrite_file_index('default.T')");
+        }
+        assertFileIndexTypes("T", "bloom-filter");
+
+        sql("ALTER TABLE T RESET ('file-index.bloom-filter.columns')");
+        sql("ALTER TABLE T SET ('file-index.bitmap.columns' = 'k')");
+        if (isNamedArgument) {
+            sql("CALL sys.rewrite_file_index(`table` => 'default.T')");
+        } else {
+            sql("CALL sys.rewrite_file_index('default.T')");
+        }
+        assertFileIndexTypes("T", "bitmap");
+    }
+
+    private void assertFileIndexTypes(String tableName, String expectedIndexType) throws Exception {
+        flinkCatalog()
+                .catalog()
+                .invalidateTable(Identifier.create(tEnv.getCurrentDatabase(), tableName));
+        FileStoreTable table = paimonTable(tableName);
+        for (ManifestEntry entry : table.store().newScan().plan().files()) {
+            byte[] embeddedIndex = entry.file().embeddedIndex();
+            FileIndexFormat.Reader reader;
+            if (embeddedIndex != null) {
+                reader =
+                        FileIndexFormat.createReader(
+                                new ByteArraySeekableStream(embeddedIndex), table.rowType());
+            } else {
+                String indexFile =
+                        entry.file().extraFiles().stream()
+                                .filter(s -> s.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX))
+                                .findFirst()
+                                .orElseThrow(
+                                        () ->
+                                                new AssertionError(
+                                                        "Missing file index for "
+                                                                + entry.file().fileName()));
+                Path indexFilePath =
+                        table.store()
+                                .pathFactory()
+                                .createDataFilePathFactory(entry.partition(), entry.bucket())
+                                .toAlignedPath(indexFile, entry.file());
+                reader =
+                        FileIndexFormat.createReader(
+                                table.fileIO().newInputStream(indexFilePath), table.rowType());
+            }
+            try (FileIndexFormat.Reader indexReader = reader) {
+                Map<String, Map<String, byte[]>> indexes = indexReader.readAll();
+                Assertions.assertThat(indexes).containsKey("k");
+                Assertions.assertThat(indexes.get("k").keySet()).containsExactly(expectedIndexType);
+            }
+        }
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void testFileIndexProcedureDropIndex(boolean isNamedArgument) throws Exception {