From 990daa65f0144e42b2723c11872aab62c8399bfa Mon Sep 17 00:00:00 2001 From: huangxiaoping <1754789345@qq.com> Date: Thu, 6 Aug 2026 23:42:50 +0800 Subject: [PATCH 1/4] [core] Fix stale file index cleanup --- .../paimon/index/FileIndexProcessor.java | 23 ++++--- .../RewriteFileIndexProcedureITCase.java | 61 +++++++++++++++++++ 2 files changed, 77 insertions(+), 7 deletions(-) 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..4fc7916f3374 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,12 @@ public DataFileMeta process(BinaryRow partition, int bucket, ManifestEntry manif maintainers.remove(name); } else { Map indexTypeBytes = maintainers.get(name); - for (String indexType : entry.getValue().keySet()) { - if (!indexTypeBytes.containsKey(indexType)) { - indexTypeBytes.remove(indexType); - } - } + Set configuredIndexTypes = + schemaInfo.projectedIndexTypes.getOrDefault( + name, Collections.emptySet()); + indexTypeBytes + .keySet() + .removeIf(indexType -> !configuredIndexTypes.contains(indexType)); } } @@ -214,6 +215,7 @@ public SchemaInfo schemaInfo(long schemaId) { List projectedColNames = new ArrayList<>(); Set projectedColFullNames = new HashSet<>(); + Map> projectedIndexTypes = new HashMap<>(); for (Map.Entry> entry : fileIndexOptions.entrySet()) { FileIndexOptions.Column column = entry.getKey(); @@ -234,6 +236,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 +249,8 @@ public SchemaInfo schemaInfo(long schemaId) { projectedColNames.stream() .mapToInt(fileSchema::getFieldIndex) .toArray(), - projectedColFullNames)); + projectedColFullNames, + projectedIndexTypes)); fileSchemaIds.add(schemaId); } @@ -276,16 +282,19 @@ private static class SchemaInfo { private final Map colNameMapping; private final int[] projectedIndexCols; private final Set projectedColFullNames; + private final Map> projectedIndexTypes; private SchemaInfo( RowType fileSchema, Map colNameMapping, int[] projectedIndexCols, - Set projectedColFullNames) { + Set projectedColFullNames, + Map> 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..2aebde10b4a3 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 @@ -37,6 +37,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 +157,66 @@ 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(paimonTable("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(paimonTable("T"), "bitmap"); + } + + private void assertFileIndexTypes(FileStoreTable table, String expectedIndexType) + throws Exception { + for (ManifestEntry entry : table.store().newScan().plan().files()) { + 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()); + try (FileIndexFormat.Reader reader = + FileIndexFormat.createReader( + table.fileIO().newInputStream(indexFilePath), table.rowType())) { + Map> indexes = reader.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 { From 8f93a7a816447a2c6c88f3acf6724cbc438e9468 Mon Sep 17 00:00:00 2001 From: huangxiaoping <1754789345@qq.com> Date: Fri, 7 Aug 2026 09:32:32 +0800 Subject: [PATCH 2/4] Fix code style --- .../main/java/org/apache/paimon/index/FileIndexProcessor.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 4fc7916f3374..325f78d28c1c 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 @@ -119,8 +119,7 @@ public DataFileMeta process(BinaryRow partition, int bucket, ManifestEntry manif } else { Map indexTypeBytes = maintainers.get(name); Set configuredIndexTypes = - schemaInfo.projectedIndexTypes.getOrDefault( - name, Collections.emptySet()); + schemaInfo.projectedIndexTypes.getOrDefault(name, Collections.emptySet()); indexTypeBytes .keySet() .removeIf(indexType -> !configuredIndexTypes.contains(indexType)); From c0098a8f03c5707b7edd824bb7c4c46039a0f3ae Mon Sep 17 00:00:00 2001 From: huangxiaoping <1754789345@qq.com> Date: Fri, 7 Aug 2026 11:34:31 +0800 Subject: [PATCH 3/4] [flink] Fix file index rewrite test cache --- .../procedure/RewriteFileIndexProcedureITCase.java | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) 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 2aebde10b4a3..ed0267a4bf80 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,6 +18,7 @@ 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; @@ -178,7 +179,7 @@ public void testFileIndexProcedureSwitchIndexType(boolean isNamedArgument) throw } else { sql("CALL sys.rewrite_file_index('default.T')"); } - assertFileIndexTypes(paimonTable("T"), "bloom-filter"); + assertFileIndexTypes("T", "bloom-filter"); sql("ALTER TABLE T RESET ('file-index.bloom-filter.columns')"); sql("ALTER TABLE T SET ('file-index.bitmap.columns' = 'k')"); @@ -187,11 +188,14 @@ public void testFileIndexProcedureSwitchIndexType(boolean isNamedArgument) throw } else { sql("CALL sys.rewrite_file_index('default.T')"); } - assertFileIndexTypes(paimonTable("T"), "bitmap"); + assertFileIndexTypes("T", "bitmap"); } - private void assertFileIndexTypes(FileStoreTable table, String expectedIndexType) - throws Exception { + 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()) { String indexFile = entry.file().extraFiles().stream() From a63ed4564d48cc89fe2b9ed3fd7662e0f57b6a44 Mon Sep 17 00:00:00 2001 From: huangxiaoping <1754789345@qq.com> Date: Fri, 7 Aug 2026 11:34:31 +0800 Subject: [PATCH 4/4] [core] Fix stale file index cleanup --- .../paimon/index/FileIndexProcessor.java | 2 +- .../RewriteFileIndexProcedureITCase.java | 46 +++++++++++-------- 2 files changed, 29 insertions(+), 19 deletions(-) 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 325f78d28c1c..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 @@ -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()); } } 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 ed0267a4bf80..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 @@ -24,6 +24,7 @@ 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; @@ -197,24 +198,33 @@ private void assertFileIndexTypes(String tableName, String expectedIndexType) th .invalidateTable(Identifier.create(tEnv.getCurrentDatabase(), tableName)); FileStoreTable table = paimonTable(tableName); for (ManifestEntry entry : table.store().newScan().plan().files()) { - 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()); - try (FileIndexFormat.Reader reader = - FileIndexFormat.createReader( - table.fileIO().newInputStream(indexFilePath), table.rowType())) { - Map> indexes = reader.readAll(); + 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> indexes = indexReader.readAll(); Assertions.assertThat(indexes).containsKey("k"); Assertions.assertThat(indexes.get("k").keySet()).containsExactly(expectedIndexType); }