Skip to content

Commit 2a5a256

Browse files
authored
[core] Avoid strict-mode conflicts for index-only compact (#8996)
1 parent 8c53b2c commit 2a5a256

2 files changed

Lines changed: 91 additions & 5 deletions

File tree

paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,13 @@ public void check(
6868
}
6969
if (snapshot.commitKind() == CommitKind.COMPACT
7070
|| snapshot.commitKind() == CommitKind.OVERWRITE) {
71-
if (hasOverlappedPartition(snapshot, newPartitions)) {
71+
boolean hasOverlap = hasOverlappedDataPartition(snapshot, newPartitions);
72+
// OVERWRITE may contain logical changes represented only by deletion vectors,
73+
// while index-only COMPACT does not change table data.
74+
if (!hasOverlap && snapshot.commitKind() == CommitKind.OVERWRITE) {
75+
hasOverlap = hasOverlappedIndexPartition(snapshot, newPartitions);
76+
}
77+
if (hasOverlap) {
7278
throw new RuntimeException(
7379
String.format(
7480
"When trying to commit snapshot %d, "
@@ -109,7 +115,7 @@ public void check(
109115
}
110116
}
111117

112-
private boolean hasOverlappedPartition(Snapshot snapshot, Set<BinaryRow> newPartitions) {
118+
private boolean hasOverlappedDataPartition(Snapshot snapshot, Set<BinaryRow> newPartitions) {
113119
if (newPartitions.isEmpty()) {
114120
return false;
115121
}
@@ -120,10 +126,13 @@ private boolean hasOverlappedPartition(Snapshot snapshot, Set<BinaryRow> newPart
120126
.withKind(ScanMode.DELTA)
121127
.dropStats()
122128
.readFileIterator();
123-
if (hasOverlappedPartition(entries, newPartitions)) {
124-
return true;
125-
}
129+
return hasOverlappedPartition(entries, newPartitions);
130+
}
126131

132+
private boolean hasOverlappedIndexPartition(Snapshot snapshot, Set<BinaryRow> newPartitions) {
133+
if (newPartitions.isEmpty()) {
134+
return false;
135+
}
127136
String indexManifest = snapshot.indexManifest();
128137
if (indexManifest == null) {
129138
return false;

paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.apache.paimon.deletionvectors.BucketedDvMaintainer;
2626
import org.apache.paimon.fs.Path;
2727
import org.apache.paimon.fs.local.LocalFileIO;
28+
import org.apache.paimon.index.GlobalIndexMeta;
2829
import org.apache.paimon.index.IndexFileHandler;
2930
import org.apache.paimon.index.IndexFileMeta;
3031
import org.apache.paimon.io.CompactIncrement;
@@ -541,6 +542,82 @@ public void testStrictModeForOverwrite() throws Exception {
541542
commit2.close();
542543
}
543544

545+
@Test
546+
public void testStrictModeIgnoresIndexOnlyCompact() throws Exception {
547+
String path = tempDir.toString();
548+
RowType rowType =
549+
RowType.of(
550+
new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.BIGINT()},
551+
new String[] {"pt", "k", "v"});
552+
553+
Options options = new Options();
554+
options.set(CoreOptions.PATH, path);
555+
options.set(CoreOptions.BUCKET, 1);
556+
options.set(CoreOptions.BUCKET_KEY, "k");
557+
options.set(CoreOptions.NUM_SORTED_RUNS_COMPACTION_TRIGGER, 10);
558+
TableSchema tableSchema =
559+
SchemaUtils.forceCommit(
560+
new SchemaManager(LocalFileIO.create(), new Path(path)),
561+
new Schema(
562+
rowType.getFields(),
563+
Collections.singletonList("pt"),
564+
Collections.emptyList(),
565+
options.toMap(),
566+
""));
567+
FileStoreTable table =
568+
FileStoreTableFactory.create(
569+
LocalFileIO.create(),
570+
new Path(path),
571+
tableSchema,
572+
CatalogEnvironment.empty());
573+
BinaryRow pt1 = partitionRow(1);
574+
575+
String user1 = UUID.randomUUID().toString();
576+
TableWriteImpl<?> write1 = table.newWrite(user1);
577+
TableCommitImpl commit1 = table.newCommit(user1);
578+
write1.write(GenericRow.of(1, 0, 0L));
579+
commit1.commit(1, write1.prepareCommit(false, 1));
580+
581+
String user2 = UUID.randomUUID().toString();
582+
FileStoreTable tableWithStrict =
583+
table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "1"));
584+
TableWriteImpl<?> write2 = tableWithStrict.newWrite(user2);
585+
TableCommitImpl commit2 = tableWithStrict.newCommit(user2);
586+
write2.write(GenericRow.of(1, 1, 1L));
587+
write2.compact(pt1, 0, true);
588+
589+
IndexFileMeta btreeIndex =
590+
new IndexFileMeta(
591+
"btree",
592+
"index-only-compact",
593+
1,
594+
1,
595+
new GlobalIndexMeta(0, 0, 1, null, null),
596+
null);
597+
commit1.commit(
598+
2,
599+
Collections.singletonList(
600+
new CommitMessageImpl(
601+
pt1,
602+
0,
603+
1,
604+
DataIncrement.emptyIncrement(),
605+
new CompactIncrement(
606+
Collections.emptyList(),
607+
Collections.emptyList(),
608+
Collections.emptyList(),
609+
Collections.singletonList(btreeIndex),
610+
Collections.emptyList()))));
611+
612+
assertThatCode(() -> commit2.commit(1, write2.prepareCommit(true, 1)))
613+
.doesNotThrowAnyException();
614+
615+
write1.close();
616+
commit1.close();
617+
write2.close();
618+
commit2.close();
619+
}
620+
544621
@Test
545622
public void testStrictModeForDvOnlyOverwrite() throws Exception {
546623
// Regression test for the partition-overlap check on DV-only OVERWRITE

0 commit comments

Comments
 (0)