Skip to content

[core] Fix stale file index cleanup #9085

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
huangxiaopingRD wants to merge 5 commits into
base: master
Choose a base branch
from
+91 −8
Open

[core] Fix stale file index cleanup #9085

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
990daa6
[core] Fix stale file index cleanup
huangxiaopingRD Aug 6, 2026
7b1a677
Merge branch 'apache:master' into fix-index-clean
huangxiaopingRD Aug 6, 2026
8f93a7a
Fix code style
huangxiaopingRD Aug 7, 2026
c0098a8
[flink] Fix file index rewrite test cache
huangxiaopingRD Aug 7, 2026
a63ed45
[core] Fix stale file index cleanup
huangxiaopingRD Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 16 additions & 8 deletions paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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));
}
}

Expand Down Expand Up @@ -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());
}
}

Expand Down Expand Up @@ -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();
Expand All @@ -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(
Expand All @@ -244,7 +248,8 @@ public SchemaInfo schemaInfo(long schemaId) {
projectedColNames.stream()
.mapToInt(fileSchema::getFieldIndex)
.toArray(),
projectedColFullNames));
projectedColFullNames,
projectedIndexTypes));
fileSchemaIds.add(schemaId);
}

Expand Down Expand Up @@ -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;
}
}
}
75 changes: 75 additions & 0 deletions ...mmon/src/test/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedureITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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 {
Expand Down
Loading