[core] Fix stale file index cleanup (#9085)
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 84e48c5..db781f9 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 @@
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 @@
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 @@
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 @@
columnName, column.getNestedColumnName())
: column.getColumnName();
projectedColFullNames.add(fullColumnName);
+ projectedIndexTypes
+ .computeIfAbsent(fullColumnName, ignored -> new HashSet<>())
+ .addAll(entry.getValue().keySet());
}
schemaInfos.put(
@@ -244,7 +248,8 @@
projectedColNames.stream()
.mapToInt(fileSchema::getFieldIndex)
.toArray(),
- projectedColFullNames));
+ projectedColFullNames,
+ projectedIndexTypes));
fileSchemaIds.add(schemaId);
}
@@ -276,16 +281,19 @@
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 fe62eb3..ad83304 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.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;
@@ -158,6 +161,130 @@
@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 {
sql(
"CREATE TABLE T ("
