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 63e77417d9 [core] Reduce index manifest memory usage during snapshot
expiration (#10710)
63e77417d9 is described below
commit 63e77417d9387e1a3c5e8a16a5e8cd76d2d0de70
Author: zhoulii <[email protected]>
AuthorDate: Wed Oct 21 13:58:52 2026 +0800
[core] Reduce index manifest memory usage during snapshot expiration
(#10710)
---
.../org/apache/paimon/index/IndexFileHandler.java | 7 +
.../index/IndexInDataFileDirPathFactory.java | 8 +-
.../org/apache/paimon/index/IndexPathFactory.java | 8 +-
.../paimon/manifest/BinaryIndexManifestEntry.java | 41 +++
.../apache/paimon/operation/ChangelogDeletion.java | 7 +-
.../apache/paimon/operation/FileDeletionBase.java | 57 ++--
.../apache/paimon/utils/FileStorePathFactory.java | 7 +-
.../manifest/BinaryIndexManifestEntryTest.java | 84 ++++++
.../operation/IndexManifestDeletionTest.java | 336 +++++++++++++++++++++
9 files changed, 522 insertions(+), 33 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
index 6621cb56cd..666384f56c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
@@ -54,6 +54,7 @@ import java.util.Set;
import java.util.function.Function;
import static org.apache.paimon.index.HashIndexFile.HASH_INDEX;
+import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
/** Handle index files. */
public class IndexFileHandler {
@@ -290,6 +291,12 @@ public class IndexFileHandler {
return pathFactories.get(entry.partition(),
entry.bucket()).toPath(entry.indexFile());
}
+ public Path filePath(BinaryIndexManifestEntry entry) {
+ return pathFactories
+ .get(deserializeBinaryRow(entry.partitionBytes()),
entry.bucket())
+ .toPath(entry.fileName(), entry.externalPath());
+ }
+
public boolean existsManifest(String indexManifest) {
return indexManifestFile.exists(indexManifest);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/IndexInDataFileDirPathFactory.java
b/paimon-core/src/main/java/org/apache/paimon/index/IndexInDataFileDirPathFactory.java
index 840d218c64..f664802164 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/IndexInDataFileDirPathFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/IndexInDataFileDirPathFactory.java
@@ -21,6 +21,8 @@ package org.apache.paimon.index;
import org.apache.paimon.fs.Path;
import org.apache.paimon.io.DataFilePathFactory;
+import javax.annotation.Nullable;
+
import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;
@@ -52,10 +54,10 @@ public class IndexInDataFileDirPathFactory implements
IndexPathFactory {
}
@Override
- public Path toPath(IndexFileMeta file) {
- return Optional.ofNullable(file.externalPath())
+ public Path toPath(String fileName, @Nullable String externalPath) {
+ return Optional.ofNullable(externalPath)
.map(Path::new)
- .orElse(new Path(dataFilePathFactory.parent(),
file.fileName()));
+ .orElse(new Path(dataFilePathFactory.parent(), fileName));
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/IndexPathFactory.java
b/paimon-core/src/main/java/org/apache/paimon/index/IndexPathFactory.java
index c1ffc75733..3180bb2c44 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/IndexPathFactory.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/IndexPathFactory.java
@@ -20,6 +20,8 @@ package org.apache.paimon.index;
import org.apache.paimon.fs.Path;
+import javax.annotation.Nullable;
+
/** Path factory to create an index path. */
public interface IndexPathFactory {
@@ -27,8 +29,12 @@ public interface IndexPathFactory {
Path newPath();
+ default Path toPath(String fileName, @Nullable String externalPath) {
+ return toPath(fileName);
+ }
+
default Path toPath(IndexFileMeta file) {
- return toPath(file.fileName());
+ return toPath(file.fileName(), file.externalPath());
}
boolean isExternalPath();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryIndexManifestEntry.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryIndexManifestEntry.java
index e951316f09..fdd3ef8de3 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryIndexManifestEntry.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryIndexManifestEntry.java
@@ -36,6 +36,18 @@ public final class BinaryIndexManifestEntry {
public static final Projection GLOBAL_INDEX_PROJECTION =
createGlobalIndexProjection();
+ public static final Projection FILE_NAME_PROJECTION =
+ Projection.create(
+
IndexManifestEntry.MANIFEST_ROW_TYPE.project(IndexManifestEntry.FILE_NAME));
+
+ public static final Projection FILE_PATH_PROJECTION =
+ Projection.create(
+ IndexManifestEntry.MANIFEST_ROW_TYPE.project(
+ IndexManifestEntry.PARTITION,
+ IndexManifestEntry.BUCKET,
+ IndexManifestEntry.FILE_NAME,
+ IndexManifestEntry.EXTERNAL_PATH));
+
private final Projection projection;
private @Nullable InternalRow row;
@@ -115,6 +127,27 @@ public final class BinaryIndexManifestEntry {
return indexType;
}
+ public String fileName() {
+ BinaryString fileName =
+ current()
+ .getString(
+ requiredPosition(
+ projection.fileNamePosition,
IndexManifestEntry.FILE_NAME));
+ checkState(fileName != null, "Index file name cannot be null.");
+ return fileName.toString();
+ }
+
+ @Nullable
+ public String externalPath() {
+ BinaryString externalPath =
+ current()
+ .getString(
+ requiredPosition(
+ projection.externalPathPosition,
+ IndexManifestEntry.EXTERNAL_PATH));
+ return externalPath == null ? null : externalPath.toString();
+ }
+
public boolean hasGlobalIndexMeta() {
return !current()
.isNullAt(
@@ -185,6 +218,8 @@ public final class BinaryIndexManifestEntry {
private final int partitionPosition;
private final int bucketPosition;
private final int indexTypePosition;
+ private final int fileNamePosition;
+ private final int externalPathPosition;
private final int globalIndexPosition;
private final int projectedGlobalIndexFieldCount;
private final int rowRangeStartPosition;
@@ -198,6 +233,8 @@ public final class BinaryIndexManifestEntry {
int partitionPosition,
int bucketPosition,
int indexTypePosition,
+ int fileNamePosition,
+ int externalPathPosition,
int globalIndexPosition,
int projectedGlobalIndexFieldCount,
int rowRangeStartPosition,
@@ -209,6 +246,8 @@ public final class BinaryIndexManifestEntry {
this.partitionPosition = partitionPosition;
this.bucketPosition = bucketPosition;
this.indexTypePosition = indexTypePosition;
+ this.fileNamePosition = fileNamePosition;
+ this.externalPathPosition = externalPathPosition;
this.globalIndexPosition = globalIndexPosition;
this.projectedGlobalIndexFieldCount =
projectedGlobalIndexFieldCount;
this.rowRangeStartPosition = rowRangeStartPosition;
@@ -246,6 +285,8 @@ public final class BinaryIndexManifestEntry {
projectedType.getFieldIndex(IndexManifestEntry.PARTITION),
projectedType.getFieldIndex(IndexManifestEntry.BUCKET),
projectedType.getFieldIndex(IndexManifestEntry.INDEX_TYPE),
+ projectedType.getFieldIndex(IndexManifestEntry.FILE_NAME),
+
projectedType.getFieldIndex(IndexManifestEntry.EXTERNAL_PATH),
globalIndexPosition,
projectedGlobalIndexFieldCount,
rowRangeStartPosition,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
index eec6090a36..24f12bed5f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
@@ -22,9 +22,7 @@ import org.apache.paimon.Changelog;
import org.apache.paimon.Snapshot;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.index.IndexFileHandler;
-import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.manifest.ExpireFileEntry;
-import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.stats.StatsFileHandler;
@@ -116,10 +114,7 @@ public class ChangelogDeletion extends
FileDeletionBase<Changelog> {
String indexManifest = skippingSnapshot.indexManifest();
if (indexManifest != null &&
indexFileHandler.existsManifest(indexManifest)) {
skippingSet.add(indexManifest);
- indexFileHandler.readManifest(indexManifest).stream()
- .map(IndexManifestEntry::indexFile)
- .map(IndexFileMeta::fileName)
- .forEach(skippingSet::add);
+ addIndexFilesToSkippingSet(skippingSnapshot, skippingSet);
}
// statistics
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
index 4f74342c0a..b449e7bdf0 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
@@ -23,20 +23,20 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.index.IndexFileHandler;
-import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.manifest.BinaryIndexManifestEntry;
import org.apache.paimon.manifest.BucketFilter;
import org.apache.paimon.manifest.ExpireFileEntry;
import org.apache.paimon.manifest.FileEntry;
import org.apache.paimon.manifest.FileEntry.Identifier;
import org.apache.paimon.manifest.FileKind;
-import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestBucketFilter;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.manifest.ProjectedManifestEntry;
import org.apache.paimon.stats.StatsFileHandler;
+import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.DataFilePathFactories;
import org.apache.paimon.utils.FileOperationThreadPool;
import org.apache.paimon.utils.FileStorePathFactory;
@@ -51,6 +51,7 @@ import javax.annotation.Nullable;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -310,24 +311,33 @@ public abstract class FileDeletionBase<T extends
Snapshot> {
private void collectUnusedIndexManifests(
Snapshot snapshot,
Set<String> skippingSet,
- Set<IndexManifestEntry> indexFiles,
+ Set<Path> indexFiles,
Set<String> indexManifests) {
// clean index manifests
String indexManifest = snapshot.indexManifest();
// check exists, it may have been deleted by other snapshots
if (indexManifest != null) {
- List<IndexManifestEntry> indexManifestEntries;
+ CloseableIterator<BinaryIndexManifestEntry> entries;
try {
- indexManifestEntries =
indexFileHandler.readManifestWithIOException(indexManifest);
- } catch (FileNotFoundException e) {
- return;
- } catch (IOException e) {
- throw new RuntimeException(e);
+ entries =
+ indexFileHandler.scan(
+ snapshot,
BinaryIndexManifestEntry.FILE_PATH_PROJECTION);
+ } catch (UncheckedIOException e) {
+ if (e.getCause() instanceof FileNotFoundException) {
+ return;
+ }
+ throw e;
}
- for (IndexManifestEntry entry : indexManifestEntries) {
- if (skippingSet.add(entry.indexFile().fileName())) {
- indexFiles.add(entry);
+ try (CloseableIterator<BinaryIndexManifestEntry> iterator =
entries) {
+ while (iterator.hasNext()) {
+ BinaryIndexManifestEntry entry = iterator.next();
+ if (skippingSet.add(entry.fileName())) {
+ indexFiles.add(indexFileHandler.filePath(entry));
+ }
}
+ } catch (Exception e) {
+ // A missing manifest is only safe to ignore before scanning
starts.
+ throw new RuntimeException("Failed to scan index manifest " +
indexManifest, e);
}
if (skippingSet.add(indexManifest)) {
@@ -363,7 +373,7 @@ public abstract class FileDeletionBase<T extends Snapshot> {
boolean deleteDataManifestLists,
boolean deleteChangelog) {
Set<String> manifests = new LinkedHashSet<>();
- Set<IndexManifestEntry> indexFiles = new LinkedHashSet<>();
+ Set<Path> indexFiles = new LinkedHashSet<>();
Set<String> indexManifests = new LinkedHashSet<>();
Set<String> statistics = new LinkedHashSet<>();
if (deleteDataManifestLists) {
@@ -392,8 +402,8 @@ public abstract class FileDeletionBase<T extends Snapshot> {
for (String manifest : manifests) {
tasks.add(() -> manifestFile.delete(manifest));
}
- for (IndexManifestEntry indexFile : indexFiles) {
- tasks.add(() -> indexFileHandler.deleteIndexFile(indexFile));
+ for (Path indexFile : indexFiles) {
+ tasks.add(() -> fileIO.deleteQuietly(indexFile));
}
for (String indexManifest : indexManifests) {
tasks.add(() -> indexFileHandler.deleteManifest(indexManifest));
@@ -618,10 +628,7 @@ public abstract class FileDeletionBase<T extends Snapshot>
{
String indexManifest = skippingSnapshot.indexManifest();
if (indexManifest != null) {
skippingSet.add(indexManifest);
- indexFileHandler.readManifest(indexManifest).stream()
- .map(IndexManifestEntry::indexFile)
- .map(IndexFileMeta::fileName)
- .forEach(skippingSet::add);
+ addIndexFilesToSkippingSet(skippingSnapshot, skippingSet);
}
String reassignPlan = planFile(skippingSnapshot);
@@ -637,6 +644,18 @@ public abstract class FileDeletionBase<T extends Snapshot>
{
return skippingSet;
}
+ protected void addIndexFilesToSkippingSet(Snapshot snapshot, Set<String>
skippingSet) {
+ try (CloseableIterator<BinaryIndexManifestEntry> entries =
+ indexFileHandler.scan(snapshot,
BinaryIndexManifestEntry.FILE_NAME_PROJECTION)) {
+ while (entries.hasNext()) {
+ skippingSet.add(entries.next().fileName());
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to scan index manifest " +
snapshot.indexManifest(), e);
+ }
+ }
+
protected static void addManifestToSkippingSet(
Set<String> skippingSet, ManifestFileMeta manifest) {
skippingSet.add(manifest.fileName());
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/FileStorePathFactory.java
b/paimon-core/src/main/java/org/apache/paimon/utils/FileStorePathFactory.java
index e2c99b52c5..981b3a24da 100644
---
a/paimon-core/src/main/java/org/apache/paimon/utils/FileStorePathFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/utils/FileStorePathFactory.java
@@ -24,7 +24,6 @@ import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.fs.ExternalPathProvider;
import org.apache.paimon.fs.Path;
-import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexInDataFileDirPathFactory;
import org.apache.paimon.index.IndexPathFactory;
import org.apache.paimon.io.ChainReadContext;
@@ -342,12 +341,12 @@ public class FileStorePathFactory {
}
@Override
- public Path toPath(IndexFileMeta file) {
- return Optional.ofNullable(file.externalPath())
+ public Path toPath(String fileName, @Nullable String externalPath)
{
+ return Optional.ofNullable(externalPath)
.map(Path::new)
// If external path is null, use the index path (not
global index root dir,
// because the root dir may change by alter table)
- .orElse(new Path(indexPath(), file.fileName()));
+ .orElse(new Path(indexPath(), fileName));
}
@Override
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryIndexManifestEntryTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryIndexManifestEntryTest.java
index 02d952ed5e..52be4fd921 100644
---
a/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryIndexManifestEntryTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryIndexManifestEntryTest.java
@@ -22,20 +22,29 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.options.MemorySize;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.TableTestBase;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.SegmentsCache;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.Arrays;
import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verifyNoInteractions;
/** Tests for {@link BinaryIndexManifestEntry}. */
class BinaryIndexManifestEntryTest extends TableTestBase {
@@ -127,6 +136,81 @@ class BinaryIndexManifestEntryTest extends TableTestBase {
assertThatThrownBy(entry::hasGlobalIndexMeta)
.isInstanceOf(UnsupportedOperationException.class)
.hasMessageContaining(IndexManifestEntry.GLOBAL_INDEX);
+ assertThatThrownBy(entry::fileName)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining(IndexManifestEntry.FILE_NAME);
+ assertThatThrownBy(entry::externalPath)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining(IndexManifestEntry.EXTERNAL_PATH);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testFileProjectionsBypassCache(boolean warmCache) throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ SegmentsCache<Path> cache =
+ spy(new SegmentsCache<>(4096, MemorySize.ofMebiBytes(8),
Long.MAX_VALUE));
+ IndexManifestFile manifest =
+ new IndexManifestFile.Factory(
+ table.fileIO(),
+ FileFormat.manifestFormat(table.coreOptions()),
+ "zstd",
+ table.store().pathFactory(),
+ cache)
+ .create();
+ IndexManifestEntry local = entry(FileKind.ADD, partition(1), 3,
"hash", null);
+ String externalPath = "file:///external/index-file";
+ IndexManifestEntry external =
+ new IndexManifestEntry(
+ FileKind.ADD,
+ partition(2),
+ 4,
+ new IndexFileMeta(
+ "btree",
+ "external-index-file",
+ 100,
+ 10,
+ new GlobalIndexMeta(0, 9, 1, null, new
byte[1024]),
+ externalPath));
+ String fileName = manifest.writeWithoutRolling(Arrays.asList(local,
external));
+ if (warmCache) {
+ assertThat(manifest.read(fileName)).containsExactly(local,
external);
+
assertThat(cache.getIfPresents(manifest.indexManifestFilePath(fileName))).isNotNull();
+ }
+ clearInvocations(cache);
+
+ try (CloseableIterator<BinaryIndexManifestEntry> entries =
+ manifest.scan(fileName,
BinaryIndexManifestEntry.FILE_PATH_PROJECTION)) {
+ BinaryIndexManifestEntry first = entries.next();
+
assertThat(first.fileName()).isEqualTo(local.indexFile().fileName());
+ assertThat(first.externalPath()).isNull();
+
assertThat(deserializeBinaryRow(first.partitionBytes())).isEqualTo(local.partition());
+ assertThat(first.bucket()).isEqualTo(local.bucket());
+ assertThatThrownBy(first::hasGlobalIndexMeta)
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ BinaryIndexManifestEntry second = entries.next();
+ assertThat(second).isSameAs(first);
+
assertThat(second.fileName()).isEqualTo(external.indexFile().fileName());
+ assertThat(second.externalPath()).isEqualTo(externalPath);
+ assertThat(deserializeBinaryRow(second.partitionBytes()))
+ .isEqualTo(external.partition());
+ assertThat(second.bucket()).isEqualTo(external.bucket());
+ assertThat(entries.hasNext()).isFalse();
+ }
+ try (CloseableIterator<BinaryIndexManifestEntry> entries =
+ manifest.scan(fileName,
BinaryIndexManifestEntry.FILE_NAME_PROJECTION)) {
+
assertThat(entries.next().fileName()).isEqualTo(local.indexFile().fileName());
+ BinaryIndexManifestEntry second = entries.next();
+
assertThat(second.fileName()).isEqualTo(external.indexFile().fileName());
+ assertThatThrownBy(second::partitionBytes)
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(second::externalPath)
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThat(entries.hasNext()).isFalse();
+ }
+ verifyNoInteractions(cache);
}
private static IndexManifestEntry entry(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/IndexManifestDeletionTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/IndexManifestDeletionTest.java
new file mode 100644
index 0000000000..4da89a1f04
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/IndexManifestDeletionTest.java
@@ -0,0 +1,336 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.TestAppendFileStore;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.manifest.BinaryIndexManifestEntry;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.IndexManifestFile;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.IndexFilePathFactories;
+import org.apache.paimon.utils.TraceableFileIO;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.paimon.data.BinaryString.fromString;
+import static org.apache.paimon.index.HashIndexFile.HASH_INDEX;
+import static
org.apache.paimon.index.IndexFileMetaSerializerTest.randomDeletionVectorIndexFile;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/** Tests for planning index deletion with projected, reusable manifest
entries. */
+class IndexManifestDeletionTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private TestAppendFileStore store;
+ private String emptyManifestList;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ store = TestAppendFileStore.createAppendStore(tempDir, new
HashMap<>());
+ emptyManifestList =
+
store.manifestListFactory().create().write(Collections.emptyList()).getLeft();
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false, false", "false, true", "true, false", "true, true"})
+ void testPlansPathsAndKeepsProtectedIndexes(boolean inDataDirectory,
boolean externalRoot)
+ throws Exception {
+ store.options()
+ .toConfiguration()
+ .set(CoreOptions.INDEX_FILE_IN_DATA_FILE_DIR, inDataDirectory);
+ if (externalRoot) {
+ store.options()
+ .toConfiguration()
+ .set(
+ CoreOptions.GLOBAL_INDEX_EXTERNAL_PATH,
+ new Path(new Path(tempDir.toUri()),
"new-index-root").toString());
+ }
+ IndexManifestFile manifest = store.indexManifestFileFactory().create();
+ IndexFileHandler handler = store.newIndexFileHandler();
+ SnapshotDeletion deletion = store.newSnapshotDeletion();
+ List<IndexManifestEntry> entries =
+ new ArrayList<>(
+ Arrays.asList(
+ hashEntry(0),
+ hashEntry(1),
+ hashEntry(2),
+ entry(3, randomDeletionVectorIndexFile()),
+ entry(
+ 4,
+ new IndexFileMeta(
+ "btree",
+ "external-index",
+ 100,
+ 10,
+ new GlobalIndexMeta(0, 9, 1,
null, new byte[1024]),
+ new Path(
+ new
Path(tempDir.toUri()),
+
"old-index-root/external-index")
+ .toString()))));
+ String protectedManifest =
+
manifest.writeWithoutRolling(Collections.singletonList(entries.get(1)));
+ String expiredManifest = manifest.writeWithoutRolling(entries);
+ String duplicateManifest = manifest.writeWithoutRolling(entries);
+ for (IndexManifestEntry entry : entries) {
+ store.fileIO().tryToWriteAtomic(handler.filePath(entry), "index");
+ }
+ Path localPath = handler.filePath(entries.get(0));
+ if (externalRoot && !inDataDirectory) {
+
assertThat(localPath.getParent()).isEqualTo(store.pathFactory().indexPath());
+ }
+
+ Snapshot protectedSnapshot = snapshot(protectedManifest);
+ Set<String> skippingSet =
+
deletion.manifestSkippingSet(Collections.singletonList(protectedSnapshot));
+ assertThat(skippingSet).contains(protectedManifest,
entries.get(1).indexFile().fileName());
+ assertThat(
+ store.newChangelogDeletion()
+
.manifestSkippingSet(Collections.singletonList(protectedSnapshot)))
+ .contains(protectedManifest,
entries.get(1).indexFile().fileName());
+ List<Runnable> tasks =
+ deletion.planManifestsCleaner(snapshot(expiredManifest),
skippingSet, false, false);
+ assertThat(tasks).hasSize(entries.size());
+ tasks.addAll(
+ deletion.planManifestsCleaner(
+ snapshot(duplicateManifest), skippingSet, false,
false));
+ assertThat(tasks).hasSize(entries.size() + 1);
+ assertThat(
+ deletion.planManifestsCleaner(
+ snapshot(expiredManifest), skippingSet, false,
false))
+ .isEmpty();
+ for (IndexManifestEntry entry : entries) {
+
assertThat(store.fileIO().exists(handler.filePath(entry))).isTrue();
+ }
+ assertThat(
+ TraceableFileIO.openInputStreams(
+ path ->
path.toString().contains(tempDir.toString())))
+ .isEmpty();
+
+ deletion.executeAll(tasks);
+
+ for (int i = 0; i < entries.size(); i++) {
+ assertThat(store.fileIO().exists(handler.filePath(entries.get(i))))
+ .as("index %s", i)
+ .isEqualTo(i == 1);
+ }
+ assertThat(handler.existsManifest(protectedManifest)).isTrue();
+ assertThat(handler.existsManifest(expiredManifest)).isFalse();
+ assertThat(handler.existsManifest(duplicateManifest)).isFalse();
+ }
+
+ @Test
+ void testMissingManifestIsIgnoredOnlyForDeletion() {
+ SnapshotDeletion deletion = store.newSnapshotDeletion();
+ assertThat(
+ deletion.planManifestsCleaner(
+ snapshot("missing"), new HashSet<>(), false,
false))
+ .isEmpty();
+ assertThatThrownBy(
+ () ->
+ deletion.manifestSkippingSet(
+
Collections.singletonList(snapshot("missing"))))
+ .hasRootCauseInstanceOf(FileNotFoundException.class);
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false, false", "false, true", "true, false", "true, true"})
+ void testScanFailureDoesNotExecutePartialPlan(boolean failOnClose, boolean
buildingSkippingSet)
+ throws Exception {
+ IndexManifestFile manifest =
spy(store.indexManifestFileFactory().create());
+ IndexFileHandler handler = spy(newHandler(manifest));
+ SnapshotDeletion deletion = newDeletion(handler);
+ IndexManifestEntry entry = hashEntry(0);
+ String fileName =
manifest.writeWithoutRolling(Collections.singletonList(entry));
+ Path indexPath = handler.filePath(entry);
+ Path manifestPath = manifest.indexManifestFilePath(fileName);
+ store.fileIO().tryToWriteAtomic(indexPath, "index");
+ BinaryIndexManifestEntry.Projection projection =
+ buildingSkippingSet
+ ? BinaryIndexManifestEntry.FILE_NAME_PROJECTION
+ : BinaryIndexManifestEntry.FILE_PATH_PROJECTION;
+ CloseableIterator<BinaryIndexManifestEntry> rows =
manifest.scan(fileName, projection);
+ FileNotFoundException failure = new FileNotFoundException("Failure
during scanning");
+ boolean[] pathPlanned = {false};
+ Set<String> skippingSet = new HashSet<>();
+ doAnswer(
+ invocation -> {
+ pathPlanned[0] = true;
+ return invocation.callRealMethod();
+ })
+ .when(handler)
+ .filePath(any(BinaryIndexManifestEntry.class));
+ CloseableIterator<BinaryIndexManifestEntry> failing =
+ new CloseableIterator<BinaryIndexManifestEntry>() {
+ private boolean consumed;
+
+ @Override
+ public boolean hasNext() {
+ if (consumed) {
+ assertThat(
+ buildingSkippingSet
+ ? skippingSet.contains(
+
entry.indexFile().fileName())
+ : pathPlanned[0])
+ .isTrue();
+ if (!failOnClose) {
+ throw new UncheckedIOException(failure);
+ }
+ }
+ return rows.hasNext();
+ }
+
+ @Override
+ public BinaryIndexManifestEntry next() {
+ consumed = true;
+ return rows.next();
+ }
+
+ @Override
+ public void close() throws Exception {
+ rows.close();
+ if (failOnClose) {
+ throw failure;
+ }
+ }
+ };
+ doReturn(failing).when(manifest).scan(fileName, projection);
+
+ assertThatThrownBy(
+ () -> {
+ if (buildingSkippingSet) {
+ deletion.addIndexFilesToSkippingSet(
+ snapshot(fileName), skippingSet);
+ } else {
+
deletion.cleanUnusedManifests(snapshot(fileName), skippingSet);
+ }
+ })
+ .hasRootCause(failure);
+ assertThat(store.fileIO().exists(indexPath)).isTrue();
+ assertThat(store.fileIO().exists(manifestPath)).isTrue();
+
assertThat(TraceableFileIO.openInputStreams(manifestPath::equals)).isEmpty();
+
+ doCallRealMethod().when(manifest).scan(fileName, projection);
+ deletion.cleanUnusedManifests(snapshot(fileName), new HashSet<>());
+ assertThat(store.fileIO().exists(indexPath)).isFalse();
+ assertThat(store.fileIO().exists(manifestPath)).isFalse();
+ }
+
+ @Test
+ void testOpeningFailureIsNotIgnored() {
+ IndexManifestFile manifest = mock(IndexManifestFile.class);
+ IOException failure = new IOException("Unable to open manifest");
+ doThrow(new UncheckedIOException(failure))
+ .when(manifest)
+ .scan("unavailable",
BinaryIndexManifestEntry.FILE_PATH_PROJECTION);
+
+ assertThatThrownBy(
+ () ->
+ newDeletion(newHandler(manifest))
+ .planManifestsCleaner(
+ snapshot("unavailable"),
+ new HashSet<>(),
+ false,
+ false))
+ .hasCause(failure);
+ }
+
+ private IndexFileHandler newHandler(IndexManifestFile manifest) {
+ return new IndexFileHandler(
+ store.fileIO(),
+ store.snapshotManager(),
+ manifest,
+ new IndexFilePathFactories(store.pathFactory()),
+ MemorySize.ofMebiBytes(2),
+ false);
+ }
+
+ private SnapshotDeletion newDeletion(IndexFileHandler handler) {
+ return new SnapshotDeletion(
+ store.fileIO(),
+ store.pathFactory(),
+ store.manifestFileFactory().create(),
+ store.manifestListFactory().create(),
+ handler,
+ store.newStatsFileHandler(),
+ false,
+ false,
+ 2,
+ null);
+ }
+
+ private Snapshot snapshot(String indexManifest) {
+ Snapshot snapshot = mock(Snapshot.class);
+ when(snapshot.baseManifestList()).thenReturn(emptyManifestList);
+ when(snapshot.baseManifestListSize()).thenReturn(null);
+ when(snapshot.deltaManifestList()).thenReturn(emptyManifestList);
+ when(snapshot.deltaManifestListSize()).thenReturn(null);
+ when(snapshot.indexManifest()).thenReturn(indexManifest);
+ return snapshot;
+ }
+
+ private IndexManifestEntry hashEntry(int partition) {
+ return entry(
+ partition,
+ new IndexFileMeta(HASH_INDEX, "hash-index-" + partition, 16,
4, null, null, null));
+ }
+
+ private IndexManifestEntry entry(int partition, IndexFileMeta indexFile) {
+ BinaryRow row = new BinaryRow(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeString(0, fromString("partition-" + partition));
+ writer.writeString(1, fromString("0"));
+ writer.complete();
+ return new IndexManifestEntry(FileKind.ADD, row, partition % 2,
indexFile);
+ }
+}
