[core] Rebuild file index over the correct column after schema evolution

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

@@ -136,9 +136,14 @@ public DataFileMeta process(BinaryRow partition, int bucket, ManifestEntry manif
                         fileIndexOptions,
                         schemaInfo.colNameMapping);
         if (dataFileIndexWriter != null) {
+            // projectedIndexCols index into the file schema. withProjection would re-interpret
+            // them against the current table schema, so a schema change that shifts columns (drop
+            // a middle column, add another) would read the wrong column and rebuild the index over
+            // it. Read with the same file-schema projection the writer above uses.
+            RowType indexReadType = schemaInfo.fileSchema.project(schemaInfo.projectedIndexCols);
             try (RecordReader<InternalRow> reader =
                     table.newReadBuilder()
-                            .withProjection(schemaInfo.projectedIndexCols)
+                            .withReadType(indexReadType)
                             .newRead()
                             .createReader(
                                     DataSplit.builder()

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

@@ -25,11 +25,14 @@
 import org.apache.paimon.data.GenericMap;
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.fileindex.FileIndexFormat;
+import org.apache.paimon.fileindex.FileIndexReader;
+import org.apache.paimon.fs.ByteArraySeekableStream;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataFilePathFactory;
 import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.predicate.FieldRef;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.table.FileStoreTable;
@@ -46,6 +49,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -174,4 +178,76 @@ public void testProcessReadsTheSchemasOfTheTableBranch() throws Exception {
         DataFileMeta processed = processor.process(entry.partition(), entry.bucket(), entry);
         assertThat(processed.extraFiles()).isNotEmpty();
     }
+
+    @Test
+    public void testRebuildsIndexOnCorrectColumnAfterColumnDropAndAdd() throws Exception {
+        LocalFileIO fileIO = LocalFileIO.create();
+        Path warehouse = new Path(tempDir.toString());
+        Map<String, String> options = new HashMap<>();
+        options.put(CoreOptions.BUCKET.key(), "1");
+        options.put(CoreOptions.FILE_FORMAT.key(), "parquet");
+        // A bitmap index is exact: querying an absent value returns an empty result,
+        // so the regression assertion below is deterministic (a bloom filter's
+        // probabilistic false positives could let the wrong-column behavior pass).
+        options.put(CoreOptions.FILE_INDEX + ".bitmap.columns", "v");
+        // v is the third field, so it sits at index 2 in the file schema.
+        RowType rowType =
+                RowType.of(
+                        new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.INT()},
+                        new String[] {"k", "a", "v"});
+
+        Identifier identifier = Identifier.create("mydb", "t");
+        try (FileSystemCatalog catalog = new FileSystemCatalog(fileIO, warehouse)) {
+            catalog.createDatabase("mydb", false);
+            catalog.createTable(
+                    identifier,
+                    new Schema(
+                            rowType.getFields(),
+                            Collections.emptyList(),
+                            Collections.singletonList("k"),
+                            options,
+                            ""),
+                    false);
+            FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+
+            String commitUser = UUID.randomUUID().toString();
+            try (TableWriteImpl<?> write = table.newWrite(commitUser);
+                    TableCommitImpl commit = table.newCommit(commitUser)) {
+                write.write(GenericRow.of(1, 10, 100));
+                commit.commit(1, write.prepareCommit(false, 1));
+            }
+
+            // Drop the middle column and add another: v keeps file-schema index 2, but the
+            // current table schema now has a different column (w) at index 2.
+            table.schemaManager().commitChanges(SchemaChange.dropColumn("a"));
+            table.schemaManager().commitChanges(SchemaChange.addColumn("w", DataTypes.INT()));
+            FileStoreTable evolved = (FileStoreTable) catalog.getTable(identifier);
+
+            List<ManifestEntry> entries = evolved.store().newScan().plan().files();
+            assertThat(entries).isNotEmpty();
+            ManifestEntry entry = entries.get(0);
+            assertThat(entry.file().schemaId()).isEqualTo(0L);
+
+            FileIndexProcessor processor = new FileIndexProcessor(evolved);
+            DataFileMeta processed = processor.process(entry.partition(), entry.bucket(), entry);
+            // The small single-row bitmap index is embedded in the manifest, not a side file.
+            byte[] embedded = processed.embeddedIndex();
+            assertThat(embedded).isNotEmpty();
+
+            // The bitmap index for v must still contain the written value 100. Before the fix
+            // the reader projected the current-schema column at index 2 (w, absent from this
+            // file), so the index was rebuilt over nulls and 100 would be absent.
+            FieldRef vRef = new FieldRef(0, "v", DataTypes.INT());
+            try (FileIndexFormat.Reader reader =
+                    FileIndexFormat.createReader(new ByteArraySeekableStream(embedded), rowType)) {
+                Set<FileIndexReader> vReaders = reader.readColumnIndex("v");
+                assertThat(vReaders).isNotEmpty();
+                for (FileIndexReader vReader : vReaders) {
+                    // Exact: 100 was indexed from v, 999 never was.
+                    assertThat(vReader.visitEqual(vRef, 100).remain()).isTrue();
+                    assertThat(vReader.visitEqual(vRef, 999).remain()).isFalse();
+                }
+            }
+        }
+    }
 }