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 de6560881655..3030ac3bb20a 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 @@ -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 reader = table.newReadBuilder() - .withProjection(schemaInfo.projectedIndexCols) + .withReadType(indexReadType) .newRead() .createReader( DataSplit.builder() diff --git a/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java b/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java index fa6f51e34d81..19a5c899e540 100644 --- a/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java +++ b/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 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 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 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(); + } + } + } + } }