This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 54e64e364c [core] Fix stale file index cleanup (#9085)
54e64e364c is described below
commit 54e64e364c1ef62206c173b4fec06a15320b0d32
Author: huangxiaoping <[email protected]>
AuthorDate: Fri Aug 14 16:29:26 2026 +0800
[core] Fix stale file index cleanup (#9085)
---
.../apache/paimon/index/FileIndexProcessor.java | 28 +++--
.../procedure/RewriteFileIndexProcedureITCase.java | 127 +++++++++++++++++++++
2 files changed, 145 insertions(+), 10 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 84e48c5538..db781f9d8a 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 class FileIndexProcessor {
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));
}
}
@@ -176,11 +176,11 @@ public class FileIndexProcessor {
outputStream.write(baos.toByteArray());
}
extras.add(newIndexPath.getName());
- return dataFileMeta.copy(extras);
+ return dataFileMeta.copy(extras).copy((byte[]) null);
} else if (baos.size() == 0) {
- return dataFileMeta.copy(extras);
+ return dataFileMeta.copy(extras).copy((byte[]) null);
} else {
- return dataFileMeta.copy(baos.toByteArray());
+ return dataFileMeta.copy(extras).copy(baos.toByteArray());
}
}
@@ -214,6 +214,7 @@ public class FileIndexProcessor {
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 class FileIndexProcessor {
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 class FileIndexProcessor {
projectedColNames.stream()
.mapToInt(fileSchema::getFieldIndex)
.toArray(),
- projectedColFullNames));
+ projectedColFullNames,
+ projectedIndexTypes));
fileSchemaIds.add(schemaId);
}
@@ -276,16 +281,19 @@ public class FileIndexProcessor {
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 fe62eb3c61..ad83304d16 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.ParameterizedTest;
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,130 @@ public class RewriteFileIndexProcedureITCase extends
CatalogITCaseBase {
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',"
+ + " 'file-index.in-manifest-threshold' = '1 MB',"
+ + " '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", true);
+
+ sql("ALTER TABLE T SET ('file-index.in-manifest-threshold' = '1 B')");
+ 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", false);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testFileIndexProcedureDropEmbeddedIndex(boolean
isNamedArgument) throws Exception {
+ sql(
+ "CREATE TABLE T ("
+ + " k INT,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'write-only' = 'true',"
+ + " 'file-index.bloom-filter.columns' = 'k',"
+ + " 'file-index.in-manifest-threshold' = '1 MB',"
+ + " 'bucket' = '-1'"
+ + ")");
+ sql("INSERT INTO T VALUES (1, '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", true);
+
+ sql("ALTER TABLE T RESET ('file-index.bloom-filter.columns')");
+ if (isNamedArgument) {
+ sql("CALL sys.rewrite_file_index(`table` => 'default.T')");
+ } else {
+ sql("CALL sys.rewrite_file_index('default.T')");
+ }
+ assertNoFileIndexes("T");
+ }
+
+ private void assertFileIndexTypes(
+ String tableName, String expectedIndexType, boolean
expectedEmbeddedIndex)
+ 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 (expectedEmbeddedIndex) {
+ Assertions.assertThat(embeddedIndex).isNotNull();
+ Assertions.assertThat(entry.file().extraFiles())
+ .noneMatch(s ->
s.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX));
+ reader =
+ FileIndexFormat.createReader(
+ new ByteArraySeekableStream(embeddedIndex),
table.rowType());
+ } else {
+ Assertions.assertThat(embeddedIndex).isNull();
+ 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);
+ }
+ }
+ }
+
+ private void assertNoFileIndexes(String tableName) throws Exception {
+ flinkCatalog()
+ .catalog()
+ .invalidateTable(Identifier.create(tEnv.getCurrentDatabase(),
tableName));
+ FileStoreTable table = paimonTable(tableName);
+ for (ManifestEntry entry : table.store().newScan().plan().files()) {
+ Assertions.assertThat(entry.file().embeddedIndex()).isNull();
+ Assertions.assertThat(entry.file().extraFiles())
+ .noneMatch(s ->
s.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX));
+ }
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testFileIndexProcedureDropIndex(boolean isNamedArgument)
throws Exception {
