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 0c10ec9124 [core] Avoid data-file conflict scans for index-only
compaction (#9963)
0c10ec9124 is described below
commit 0c10ec912484a865f43cc38120f7f2d9f40cd887
Author: YeJunHao <[email protected]>
AuthorDate: Sun Sep 20 11:09:54 2026 +0800
[core] Avoid data-file conflict scans for index-only compaction (#9963)
---
.../paimon/operation/FileStoreCommitImpl.java | 4 +-
.../paimon/operation/commit/ConflictDetection.java | 8 +
.../commit/DataEvolutionConflictDetection.java | 107 ++++++++
.../operation/commit/ConflictDetectionTest.java | 275 +++++++++++++++++++++
.../table/DataEvolutionDeletionVectorTest.java | 61 +++++
5 files changed, 454 insertions(+), 1 deletion(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index e867a1cc85..93092e20d7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -1045,7 +1045,9 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
if (!checkConflicts && shouldCheckSameFixedBucket(commitKind)) {
checkSameFixedBucketFromSnapshot(deltaFiles, latestSnapshot);
}
- if (checkConflicts) {
+ if (checkConflicts
+ && !conflictDetection.canSkipDataFileConflictDetection(
+ latestSnapshot, deltaFiles, indexFiles, commitKind)) {
// latestSnapshotId is different from the snapshot id we've
checked for conflicts,
// so we have to check again
if (changedPartitions == null) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
index 1a202ac95e..212ce0962a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
@@ -288,6 +288,14 @@ public abstract class ConflictDetection {
CommitKind commitKind,
String baseCommitUser);
+ public boolean canSkipDataFileConflictDetection(
+ Snapshot latestSnapshot,
+ List<ManifestEntry> deltaFiles,
+ List<IndexManifestEntry> indexFiles,
+ CommitKind commitKind) {
+ return false;
+ }
+
public List<SimpleFileEntry> scanBaseDataFiles(
Snapshot latestSnapshot,
List<BinaryRow> changedPartitions,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
index 056c6b8dca..c1a31e1de3 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
@@ -51,6 +51,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -72,6 +73,7 @@ public class DataEvolutionConflictDetection extends
ConflictDetection {
private final String tableName;
private final String commitUser;
private final SnapshotManager snapshotManager;
+ private final IndexFileHandler indexFileHandler;
private final boolean nestedFieldEnabled;
private @Nullable Long rowIdCheckFromSnapshot;
@@ -100,6 +102,7 @@ public class DataEvolutionConflictDetection extends
ConflictDetection {
this.tableName = tableName;
this.commitUser = commitUser;
this.snapshotManager = snapshotManager;
+ this.indexFileHandler = indexFileHandler;
this.nestedFieldEnabled = nestedFieldEnabled;
}
@@ -145,6 +148,110 @@ public class DataEvolutionConflictDetection extends
ConflictDetection {
return rowIdConflictCheckStrategy;
}
+ @Override
+ public boolean canSkipDataFileConflictDetection(
+ Snapshot latestSnapshot,
+ List<ManifestEntry> deltaFiles,
+ List<IndexManifestEntry> indexFiles,
+ CommitKind commitKind) {
+ if (commitKind != CommitKind.COMPACT
+ || !deltaFiles.isEmpty()
+ || indexFiles.isEmpty()
+ || rowIdCheckFromSnapshot != null
+ || latestSnapshot.indexManifest() == null) {
+ return false;
+ }
+
+ Map<String, IndexManifestEntry> removedFiles = new HashMap<>();
+ Set<String> addedFiles = new HashSet<>();
+ Map<IndexCompactionKey, List<Range>> removedRanges = new HashMap<>();
+ Map<IndexCompactionKey, List<Range>> addedRanges = new HashMap<>();
+ for (IndexManifestEntry entry : indexFiles) {
+ IndexFileMeta file = entry.indexFile();
+ GlobalIndexMeta meta = file.globalIndexMeta();
+ if (meta == null || file.dvRanges() != null) {
+ return false;
+ }
+ Map<IndexCompactionKey, List<Range>> ranges;
+ if (entry.kind() == FileKind.DELETE) {
+ if (removedFiles.put(file.fileName(), entry) != null) {
+ return false;
+ }
+ ranges = removedRanges;
+ } else {
+ if (!addedFiles.add(file.fileName())) {
+ return false;
+ }
+ ranges = addedRanges;
+ }
+ ranges.computeIfAbsent(new IndexCompactionKey(entry), key -> new
ArrayList<>())
+ .add(meta.rowRange());
+ }
+ if (removedFiles.isEmpty()
+ || addedFiles.isEmpty()
+ || !Collections.disjoint(removedFiles.keySet(), addedFiles)
+ || !removedRanges.keySet().equals(addedRanges.keySet())) {
+ return false;
+ }
+ for (Map.Entry<IndexCompactionKey, List<Range>> removed :
removedRanges.entrySet()) {
+ if (!Range.sortAndMergeOverlap(removed.getValue(), true)
+
.equals(Range.sortAndMergeOverlap(addedRanges.get(removed.getKey()), true))) {
+ return false;
+ }
+ }
+
+ for (IndexManifestEntry current :
+ indexFileHandler.readManifest(latestSnapshot.indexManifest()))
{
+ IndexManifestEntry removed =
removedFiles.get(current.indexFile().fileName());
+ if (removed == null) {
+ continue;
+ }
+ if (current.kind() != FileKind.ADD
+ || !current.partition().equals(removed.partition())
+ || current.bucket() != removed.bucket()
+ || !current.indexFile().equals(removed.indexFile())) {
+ return false;
+ }
+ removedFiles.remove(current.indexFile().fileName());
+ }
+ return removedFiles.isEmpty();
+ }
+
+ private static class IndexCompactionKey {
+
+ private final BinaryRow partition;
+ private final int bucket;
+ private final String indexType;
+ private final List<Integer> fieldIds;
+
+ private IndexCompactionKey(IndexManifestEntry entry) {
+ this.partition = entry.partition();
+ this.bucket = entry.bucket();
+ this.indexType = entry.indexFile().indexType();
+ this.fieldIds =
entry.indexFile().globalIndexMeta().getIndexedFieldIds();
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+ if (!(other instanceof IndexCompactionKey)) {
+ return false;
+ }
+ IndexCompactionKey that = (IndexCompactionKey) other;
+ return bucket == that.bucket
+ && partition.equals(that.partition)
+ && indexType.equals(that.indexType)
+ && fieldIds.equals(that.fieldIds);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(partition, bucket, indexType, fieldIds);
+ }
+ }
+
@Override
public List<SimpleFileEntry> scanBaseDataFiles(
Snapshot latestSnapshot,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
index 787d0761ea..41022f77b4 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.index.DeletionVectorMeta;
import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileEntry;
@@ -1560,6 +1561,280 @@ class ConflictDetectionTest {
return (DataEvolutionConflictDetection) createConflictDetection(null,
true, false);
}
+ @Test
+ void testIndexOnlyCompactionSkipsDataFileConflictDetection() {
+ IndexFileHandler handler = mock(IndexFileHandler.class);
+ CommitScanner scanner = mock(CommitScanner.class);
+ DataEvolutionConflictDetection detection =
indexCompactionDetection(handler, scanner);
+ Snapshot snapshot = indexSnapshot("indexes");
+ IndexManifestEntry first = createGlobalIndexEntry("first", ADD,
EMPTY_ROW, 0, 49);
+ IndexManifestEntry second = createGlobalIndexEntry("second", ADD,
EMPTY_ROW, 50, 99);
+ when(handler.readManifest("indexes"))
+ .thenReturn(
+ Arrays.asList(
+ first,
+ second,
+ createGlobalIndexEntry("retained", ADD,
EMPTY_ROW, 100, 199)));
+
+ assertThat(
+ detection.canSkipDataFileConflictDetection(
+ snapshot,
+ Collections.emptyList(),
+ Arrays.asList(
+ first.toDeleteEntry(),
+ second.toDeleteEntry(),
+ createGlobalIndexEntry("output-1",
ADD, EMPTY_ROW, 0, 99),
+ createGlobalIndexEntry("output-2",
ADD, EMPTY_ROW, 0, 99)),
+ Snapshot.CommitKind.COMPACT))
+ .isTrue();
+ verify(handler).readManifest("indexes");
+ verifyNoInteractions(scanner);
+ }
+
+ @Test
+ void testIndexOnlyCompactionRechecksInputsForEverySnapshot() {
+ IndexFileHandler handler = mock(IndexFileHandler.class);
+ DataEvolutionConflictDetection detection =
indexCompactionDetection(handler, null);
+ IndexManifestEntry input = createGlobalIndexEntry("input", ADD,
EMPTY_ROW, 0, 99);
+ List<IndexManifestEntry> changes =
+ Arrays.asList(
+ input.toDeleteEntry(),
+ createGlobalIndexEntry("output", ADD, EMPTY_ROW, 0,
99));
+
when(handler.readManifest("before")).thenReturn(Collections.singletonList(input));
+
when(handler.readManifest("after-append")).thenReturn(Collections.singletonList(input));
+ when(handler.readManifest("after-reassign"))
+ .thenReturn(
+ Collections.singletonList(
+ createGlobalIndexEntry("input", ADD,
EMPTY_ROW, 1000, 1099)));
+ when(handler.readManifest("after-replacement"))
+ .thenReturn(
+ Collections.singletonList(
+ createGlobalIndexEntry("other", ADD,
EMPTY_ROW, 0, 99)));
+
+ for (String manifest : Arrays.asList("before", "after-append")) {
+ assertThat(
+ detection.canSkipDataFileConflictDetection(
+ indexSnapshot(manifest),
+ Collections.emptyList(),
+ changes,
+ Snapshot.CommitKind.COMPACT))
+ .isTrue();
+ }
+ for (String manifest : Arrays.asList("after-reassign",
"after-replacement")) {
+ assertThat(
+ detection.canSkipDataFileConflictDetection(
+ indexSnapshot(manifest),
+ Collections.emptyList(),
+ changes,
+ Snapshot.CommitKind.COMPACT))
+ .isFalse();
+ }
+ }
+
+ @Test
+ void testIndexOnlyCompactionRequiresMatchingInputMetadata() {
+ IndexFileHandler handler = mock(IndexFileHandler.class);
+ DataEvolutionConflictDetection detection =
indexCompactionDetection(handler, null);
+ IndexManifestEntry input = createGlobalIndexEntry("input", ADD,
EMPTY_ROW, 0, 99);
+ List<IndexManifestEntry> changes =
+ Arrays.asList(
+ input.toDeleteEntry(),
+ createGlobalIndexEntry("output", ADD, EMPTY_ROW, 0,
99));
+ List<IndexManifestEntry> changedInputs =
+ Arrays.asList(
+ new IndexManifestEntry(
+ ADD, BinaryRow.singleColumn(1), 0,
input.indexFile()),
+ new IndexManifestEntry(ADD, EMPTY_ROW, 1,
input.indexFile()),
+ input.toDeleteEntry(),
+ indexEntryWithMeta(
+ "input", "btree", new GlobalIndexMeta(0, 99,
1, null, null)),
+ indexEntryWithMeta(
+ "input",
+ "btree",
+ new GlobalIndexMeta(0, 99, 0, new int[] {1},
null)),
+ indexEntryWithMeta(
+ "input",
+ "btree",
+ new GlobalIndexMeta(0, 99, 0, null, new byte[]
{1})),
+ indexEntryWithMeta(
+ "input",
+ "btree",
+ new GlobalIndexMeta(0, 99, 0, null, null, new
byte[] {1})),
+ indexEntryWithMeta(
+ "input", "bitmap", new GlobalIndexMeta(0, 99,
0, null, null)));
+ for (IndexManifestEntry changedInput : changedInputs) {
+ when(handler.readManifest("indexes"))
+ .thenReturn(Collections.singletonList(changedInput));
+ assertThat(
+ detection.canSkipDataFileConflictDetection(
+ indexSnapshot("indexes"),
+ Collections.emptyList(),
+ changes,
+ Snapshot.CommitKind.COMPACT))
+ .as("changed input: %s", changedInput)
+ .isFalse();
+ }
+ }
+
+ @Test
+ void testIndexOnlyCompactionRequiresMatchingOutputScope() {
+ IndexFileHandler handler = mock(IndexFileHandler.class);
+ DataEvolutionConflictDetection detection =
indexCompactionDetection(handler, null);
+ IndexManifestEntry input = createGlobalIndexEntry("input", ADD,
EMPTY_ROW, 0, 99);
+ IndexManifestEntry output = createGlobalIndexEntry("output", ADD,
EMPTY_ROW, 0, 99);
+ List<IndexManifestEntry> changedOutputs =
+ Arrays.asList(
+ new IndexManifestEntry(
+ ADD, BinaryRow.singleColumn(1), 0,
output.indexFile()),
+ new IndexManifestEntry(ADD, EMPTY_ROW, 1,
output.indexFile()),
+ indexEntryWithMeta(
+ "output", "bitmap", new GlobalIndexMeta(0, 99,
0, null, null)),
+ indexEntryWithMeta(
+ "output", "btree", new GlobalIndexMeta(0, 99,
1, null, null)),
+ indexEntryWithMeta(
+ "output",
+ "btree",
+ new GlobalIndexMeta(0, 99, 0, new int[] {1},
null)));
+ for (IndexManifestEntry changedOutput : changedOutputs) {
+ assertThat(
+ detection.canSkipDataFileConflictDetection(
+ indexSnapshot("indexes"),
+ Collections.emptyList(),
+ Arrays.asList(input.toDeleteEntry(),
changedOutput),
+ Snapshot.CommitKind.COMPACT))
+ .isFalse();
+ }
+ verifyNoInteractions(handler);
+ }
+
+ @Test
+ void testIndexOnlyCompactionDoesNotFillGapsOrChangeCoverage() {
+ IndexFileHandler handler = mock(IndexFileHandler.class);
+ DataEvolutionConflictDetection detection =
indexCompactionDetection(handler, null);
+ IndexManifestEntry first = createGlobalIndexEntry("first", DELETE,
EMPTY_ROW, 0, 49);
+ IndexManifestEntry second = createGlobalIndexEntry("second", DELETE,
EMPTY_ROW, 60, 99);
+ for (Range output : Arrays.asList(new Range(0, 99), new Range(0, 49),
new Range(0, 109))) {
+ assertThat(
+ detection.canSkipDataFileConflictDetection(
+ indexSnapshot("indexes"),
+ Collections.emptyList(),
+ Arrays.asList(
+ first,
+ second,
+ createGlobalIndexEntry(
+ "output",
+ ADD,
+ EMPTY_ROW,
+ output.from,
+ output.to)),
+ Snapshot.CommitKind.COMPACT))
+ .isFalse();
+ }
+ verifyNoInteractions(handler);
+ }
+
+ @Test
+ void testIndexOnlyCompactionFallsBackForOtherIndexChanges() {
+ IndexFileHandler handler = mock(IndexFileHandler.class);
+ DataEvolutionConflictDetection detection =
indexCompactionDetection(handler, null);
+ IndexManifestEntry input = createGlobalIndexEntry("input", DELETE,
EMPTY_ROW, 0, 99);
+ IndexManifestEntry output = createGlobalIndexEntry("output", ADD,
EMPTY_ROW, 0, 99);
+ List<List<IndexManifestEntry>> changes =
+ Arrays.asList(
+ Collections.emptyList(),
+ Collections.singletonList(input),
+ Collections.singletonList(output),
+ Arrays.asList(input, input, output),
+ Arrays.asList(input, output, output),
+ Arrays.asList(
+ input, createGlobalIndexEntry("input", ADD,
EMPTY_ROW, 0, 99)),
+ Arrays.asList(
+ input,
+ output,
+ createDvIndexEntry("dv", ADD,
Collections.singletonList("data"))));
+ for (List<IndexManifestEntry> change : changes) {
+ assertThat(
+ detection.canSkipDataFileConflictDetection(
+ indexSnapshot("indexes"),
+ Collections.emptyList(),
+ change,
+ Snapshot.CommitKind.COMPACT))
+ .isFalse();
+ }
+ verifyNoInteractions(handler);
+ }
+
+ @Test
+ void testIndexOnlyCompactionDoesNotBypassDataOrHistoricalChecks() {
+ IndexFileHandler handler = mock(IndexFileHandler.class);
+ DataEvolutionConflictDetection detection =
indexCompactionDetection(handler, null);
+ List<IndexManifestEntry> changes =
+ Arrays.asList(
+ createGlobalIndexEntry("input", DELETE, EMPTY_ROW, 0,
99),
+ createGlobalIndexEntry("output", ADD, EMPTY_ROW, 0,
99));
+ for (Snapshot.CommitKind kind :
+ Arrays.asList(Snapshot.CommitKind.APPEND,
Snapshot.CommitKind.OVERWRITE)) {
+ assertThat(
+ detection.canSkipDataFileConflictDetection(
+ indexSnapshot("indexes"),
+ Collections.emptyList(),
+ changes,
+ kind))
+ .isFalse();
+ }
+ assertThat(
+ detection.canSkipDataFileConflictDetection(
+ indexSnapshot("indexes"),
+ Collections.singletonList(
+ manifestEntry(ADD, "data", 0L, new
Range(0, 99))),
+ changes,
+ Snapshot.CommitKind.COMPACT))
+ .isFalse();
+ assertThat(
+ detection.canSkipDataFileConflictDetection(
+ snapshot(1),
+ Collections.emptyList(),
+ changes,
+ Snapshot.CommitKind.COMPACT))
+ .isFalse();
+ detection.setRowIdCheckFromSnapshot(1L);
+ assertThat(
+ detection.canSkipDataFileConflictDetection(
+ indexSnapshot("indexes"),
+ Collections.emptyList(),
+ changes,
+ Snapshot.CommitKind.COMPACT))
+ .isFalse();
+ verifyNoInteractions(handler);
+ }
+
+ private DataEvolutionConflictDetection indexCompactionDetection(
+ IndexFileHandler handler, @Nullable CommitScanner scanner) {
+ return new DataEvolutionConflictDetection(
+ "test-table",
+ "test-user",
+ RowType.of(),
+ null,
+ BucketMode.BUCKET_UNAWARE,
+ false,
+ false,
+ handler,
+ null,
+ scanner);
+ }
+
+ private Snapshot indexSnapshot(String manifest) {
+ Snapshot snapshot = mock(Snapshot.class);
+ when(snapshot.indexManifest()).thenReturn(manifest);
+ return snapshot;
+ }
+
+ private IndexManifestEntry indexEntryWithMeta(
+ String fileName, String indexType, GlobalIndexMeta meta) {
+ return new IndexManifestEntry(
+ ADD, EMPTY_ROW, 0, new IndexFileMeta(indexType, fileName, 11,
1, meta, null));
+ }
+
private ConflictDetection createConflictDetection(
@Nullable CommitScanner scanner,
boolean dataEvolutionEnabled,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
index 8f189a7917..ef7eef3196 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
@@ -36,6 +36,9 @@ import org.apache.paimon.deletionvectors.BitmapDeletionVector;
import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
import org.apache.paimon.format.blob.BlobFileFormat;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.index.DeletionVectorMeta;
import org.apache.paimon.index.GlobalIndexMeta;
@@ -46,6 +49,7 @@ import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
@@ -60,6 +64,7 @@ import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.EndOfScanException;
import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.types.DataTypes;
@@ -80,6 +85,7 @@ import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
@@ -391,6 +397,61 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
assertThat(readPartitionedRowsWithRowIds(table)).isEqualTo(reassignedRows);
}
+ @Test
+ public void testGlobalIndexCompactionDoesNotReadDataManifests() throws
Exception {
+ FileStoreTable table =
+ createPartitionedReassignTable("index_only_compaction", false)
+ .copy(
+ Collections.singletonMap(
+
CoreOptions.MANIFEST_MERGE_MIN_COUNT.key(),
+ Integer.toString(Integer.MAX_VALUE)));
+ writePartitionRows(table, "a", 0, 1);
+ BinaryRow partitionA = partition(table, "a");
+ IndexFileMeta first = globalIndexFile("first-index", new Range(0, 0));
+ IndexFileMeta second = globalIndexFile("second-index", new Range(1,
1));
+ commitGlobalIndex(table, partitionA, first);
+ commitGlobalIndex(table, partitionA, second);
+ IndexFileMeta compacted = globalIndexFile("compacted-index", new
Range(0, 1));
+ CommitMessage compaction =
+ new CommitMessageImpl(
+ partitionA,
+ UNAWARE_BUCKET,
+ null,
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.singletonList(compacted),
+ Arrays.asList(first, second)));
+
+ writePartitionRows(table, "a", 2);
+ Snapshot before = table.latestSnapshot().get();
+ Set<String> dataManifests =
+ table.store().newScan().manifestsReader().read(before,
ScanMode.ALL)
+ .filteredManifests.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toSet());
+ assertThat(dataManifests).isNotEmpty();
+ LocalFileIO guardedFileIO =
+ new LocalFileIO() {
+ @Override
+ public SeekableInputStream newInputStream(Path path)
throws IOException {
+
assertThat(dataManifests).doesNotContain(path.getName());
+ return super.newInputStream(path);
+ }
+ };
+ FileStoreTable guardedTable =
+ FileStoreTableFactory.create(guardedFileIO, table.location(),
table.schema());
+
+ commit(guardedTable, Collections.singletonList(compaction));
+
+ assertThat(table.latestSnapshot().get().id()).isEqualTo(before.id() +
1);
+ assertThat(table.latestSnapshot().get().commitKind())
+ .isEqualTo(Snapshot.CommitKind.COMPACT);
+
assertThat(liveGlobalIndexFileNames(table)).containsExactly("compacted-index");
+ }
+
@Test
public void testStaleGlobalIndexCompactIsRejectedAfterRowIdReassign()
throws Exception {
FileStoreTable table =
