Skip to content

Commit ab05eba

Browse files
yuzelinyuzelin
authored
[core] StrictModeChecker should also check index (#7884)
1 parent 8bb3816 commit ab05eba

3 files changed

Lines changed: 217 additions & 2 deletions

File tree

‎paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -210,7 +210,11 @@ public FileStoreCommitImpl(
210210
.map(
211211
id ->
212212
new StrictModeChecker(
213-
snapshotManager, commitUser, scanSupplier, id))
213+
snapshotManager,
214+
commitUser,
215+
scanSupplier,
216+
indexManifestFile,
217+
id))
214218
.orElse(null);
215219
this.conflictDetection = conflictDetectFactory.create(scanner);
216220
this.commitCleaner = new CommitCleaner(manifestList, manifestFile, indexManifestFile);

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

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222
import org.apache.paimon.Snapshot;
2323
import org.apache.paimon.Snapshot.CommitKind;
2424
import org.apache.paimon.data.BinaryRow;
25+
import org.apache.paimon.manifest.IndexManifestEntry;
26+
import org.apache.paimon.manifest.IndexManifestFile;
2527
import org.apache.paimon.manifest.ManifestEntry;
2628
import org.apache.paimon.operation.FileStoreScan;
2729
import org.apache.paimon.table.source.ScanMode;
@@ -39,17 +41,20 @@ public class StrictModeChecker {
3941
private final SnapshotManager snapshotManager;
4042
private final String commitUser;
4143
private final Supplier<FileStoreScan> scanSupplier;
44+
private final IndexManifestFile indexManifestFile;
4245

4346
private long strictModeLastSafeSnapshot;
4447

4548
public StrictModeChecker(
4649
SnapshotManager snapshotManager,
4750
String commitUser,
4851
Supplier<FileStoreScan> scanSupplier,
52+
IndexManifestFile indexManifestFile,
4953
long strictModeLastSafeSnapshot) {
5054
this.snapshotManager = snapshotManager;
5155
this.commitUser = commitUser;
5256
this.scanSupplier = scanSupplier;
57+
this.indexManifestFile = indexManifestFile;
5358
this.strictModeLastSafeSnapshot = strictModeLastSafeSnapshot;
5459
}
5560

@@ -115,7 +120,28 @@ private boolean hasOverlappedPartition(Snapshot snapshot, Set<BinaryRow> newPart
115120
.withKind(ScanMode.DELTA)
116121
.dropStats()
117122
.readFileIterator();
118-
return hasOverlappedPartition(entries, newPartitions);
123+
if (hasOverlappedPartition(entries, newPartitions)) {
124+
return true;
125+
}
126+
127+
String indexManifest = snapshot.indexManifest();
128+
if (indexManifest == null) {
129+
return false;
130+
}
131+
// Fast exit: if this snapshot's indexManifest file name equals the
132+
// previous snapshot's, no index file was added/removed by this commit
133+
// (writeIndexFiles reuses the previous file when newIndexFiles is empty).
134+
long prevId = snapshot.id() - 1;
135+
if (snapshotManager.snapshotExists(prevId)
136+
&& indexManifest.equals(snapshotManager.snapshot(prevId).indexManifest())) {
137+
return false;
138+
}
139+
for (IndexManifestEntry entry : indexManifestFile.read(indexManifest)) {
140+
if (newPartitions.contains(entry.partition())) {
141+
return true;
142+
}
143+
}
144+
return false;
119145
}
120146

121147
private boolean hasOverlappedPartition(

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

Lines changed: 185 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,11 @@
2222
import org.apache.paimon.data.BinaryRow;
2323
import org.apache.paimon.data.BinaryRowWriter;
2424
import org.apache.paimon.data.GenericRow;
25+
import org.apache.paimon.deletionvectors.BucketedDvMaintainer;
2526
import org.apache.paimon.fs.Path;
2627
import org.apache.paimon.fs.local.LocalFileIO;
28+
import org.apache.paimon.index.IndexFileHandler;
29+
import org.apache.paimon.index.IndexFileMeta;
2730
import org.apache.paimon.io.CompactIncrement;
2831
import org.apache.paimon.io.DataFileMeta;
2932
import org.apache.paimon.io.DataFilePathFactory;
@@ -538,6 +541,188 @@ public void testStrictModeForOverwrite() throws Exception {
538541
commit2.close();
539542
}
540543

544+
@Test
545+
public void testStrictModeForDvOnlyOverwrite() throws Exception {
546+
// Regression test for the partition-overlap check on DV-only OVERWRITE
547+
// snapshots: such snapshots have no data-file delta, so the check must
548+
// also look at index-manifest delta.
549+
String path = tempDir.toString();
550+
RowType rowType =
551+
RowType.of(
552+
new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.BIGINT()},
553+
new String[] {"pt", "k", "v"});
554+
555+
Options options = new Options();
556+
options.set(CoreOptions.PATH, path);
557+
options.set(CoreOptions.BUCKET, 1);
558+
options.set(CoreOptions.BUCKET_KEY, "k");
559+
options.set(CoreOptions.NUM_SORTED_RUNS_COMPACTION_TRIGGER, 10);
560+
options.set(CoreOptions.DELETION_VECTORS_ENABLED, true);
561+
TableSchema tableSchema =
562+
SchemaUtils.forceCommit(
563+
new SchemaManager(LocalFileIO.create(), new Path(path)),
564+
new Schema(
565+
rowType.getFields(),
566+
Collections.singletonList("pt"),
567+
Collections.emptyList(),
568+
options.toMap(),
569+
""));
570+
FileStoreTable table =
571+
FileStoreTableFactory.create(
572+
LocalFileIO.create(),
573+
new Path(path),
574+
tableSchema,
575+
CatalogEnvironment.empty());
576+
BinaryRow pt1 = partitionRow(1);
577+
BinaryRow pt2 = partitionRow(2);
578+
579+
// user1 writes pt=1 and pt=2 -> snapshot 1 (APPEND)
580+
String user1 = UUID.randomUUID().toString();
581+
TableWriteImpl<?> write1 = table.newWrite(user1);
582+
TableCommitImpl commit1 = table.newCommit(user1);
583+
write1.write(GenericRow.of(1, 0, 0L));
584+
write1.write(GenericRow.of(2, 0, 0L));
585+
commit1.commit(1, write1.prepareCommit(false, 1));
586+
587+
// user2 with strict mode, last-safe-snapshot=2 (skip its own snapshot 2)
588+
String user2 = UUID.randomUUID().toString();
589+
FileStoreTable tableWithStrict =
590+
table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "2"));
591+
TableWriteImpl<?> write2 = tableWithStrict.newWrite(user2);
592+
TableCommitImpl commit2 = tableWithStrict.newCommit(user2);
593+
write2.write(GenericRow.of(1, 1, 1L));
594+
commit2.commit(1, write2.prepareCommit(false, 1));
595+
596+
// user1 DV-only OVERWRITE on pt=2 -> snapshot 3 (OVERWRITE, no data-file delta)
597+
commitDvOnly(table, user1, pt2, 2);
598+
599+
// user2 COMPACT on pt=1: DV-only on pt=2 does not overlap -> no error
600+
write2.write(GenericRow.of(1, 5, 5L));
601+
write2.compact(pt1, 0, true);
602+
assertThatCode(() -> commit2.commit(2, write2.prepareCommit(true, 2)))
603+
.doesNotThrowAnyException();
604+
605+
// user1 DV-only OVERWRITE on pt=1 -> snapshot 5 (OVERWRITE, no data-file delta)
606+
commitDvOnly(table, user1, pt1, 3);
607+
608+
// user2 COMPACT on pt=1: DV-only on pt=1 overlaps -> must throw
609+
write2.write(GenericRow.of(1, 7, 7L));
610+
write2.compact(pt1, 0, true);
611+
assertThatThrownBy(() -> commit2.commit(3, write2.prepareCommit(true, 3)))
612+
.isInstanceOf(RuntimeException.class)
613+
.hasMessageContaining(
614+
"Giving up committing as commit.strict-mode.last-safe-snapshot is set.");
615+
616+
write1.close();
617+
commit1.close();
618+
write2.close();
619+
commit2.close();
620+
}
621+
622+
@Test
623+
public void testStrictModeShortCircuitsWhenIndexManifestUnchanged() throws Exception {
624+
String path = tempDir.toString();
625+
RowType rowType =
626+
RowType.of(
627+
new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.BIGINT()},
628+
new String[] {"pt", "k", "v"});
629+
630+
Options options = new Options();
631+
options.set(CoreOptions.PATH, path);
632+
options.set(CoreOptions.BUCKET, 1);
633+
options.set(CoreOptions.BUCKET_KEY, "k");
634+
options.set(CoreOptions.NUM_SORTED_RUNS_COMPACTION_TRIGGER, 10);
635+
options.set(CoreOptions.DELETION_VECTORS_ENABLED, true);
636+
TableSchema tableSchema =
637+
SchemaUtils.forceCommit(
638+
new SchemaManager(LocalFileIO.create(), new Path(path)),
639+
new Schema(
640+
rowType.getFields(),
641+
Collections.singletonList("pt"),
642+
Collections.emptyList(),
643+
options.toMap(),
644+
""));
645+
FileStoreTable table =
646+
FileStoreTableFactory.create(
647+
LocalFileIO.create(),
648+
new Path(path),
649+
tableSchema,
650+
CatalogEnvironment.empty());
651+
BinaryRow pt3 = partitionRow(3);
652+
653+
// user1 writes pt=1 and pt=3 -> snapshot 1 (APPEND)
654+
String user1 = UUID.randomUUID().toString();
655+
TableWriteImpl<?> write1 = table.newWrite(user1);
656+
TableCommitImpl commit1 = table.newCommit(user1);
657+
write1.write(GenericRow.of(1, 0, 0L));
658+
write1.write(GenericRow.of(3, 0, 0L));
659+
commit1.commit(1, write1.prepareCommit(false, 1));
660+
661+
// user1 commits a DV on pt=3 -> snapshot 2 (OVERWRITE, dv-only).
662+
// indexManifest now records pt=3.
663+
commitDvOnly(table, user1, pt3, 2);
664+
665+
// user1 OVERWRITE on pt=1 -> snapshot 3 (OVERWRITE).
666+
// pt=1 has no DV, so writeIndexFiles produces no new index file and
667+
// snapshot 3 reuses snapshot 2's indexManifest file name.
668+
TableCommitImpl commit1Ow = table.newCommit(user1).withOverwrite(singletonMap("pt", "1"));
669+
write1.write(GenericRow.of(1, 9, 9L));
670+
commit1Ow.commit(3, write1.prepareCommit(false, 3));
671+
672+
// user2 with last-safe=2 writes pt=3 -> APPEND.
673+
// Snapshot 3 is OVERWRITE on pt=1, its data delta does not touch pt=3,
674+
// and although its indexManifest still contains pt=3 (inherited from
675+
// snapshot 2), the short-circuit must skip the full-set intersection
676+
// and avoid throwing.
677+
String user2 = UUID.randomUUID().toString();
678+
FileStoreTable tableWithStrict =
679+
table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "2"));
680+
TableWriteImpl<?> write2 = tableWithStrict.newWrite(user2);
681+
TableCommitImpl commit2 = tableWithStrict.newCommit(user2);
682+
write2.write(GenericRow.of(3, 1, 1L));
683+
assertThatCode(() -> commit2.commit(1, write2.prepareCommit(false, 1)))
684+
.doesNotThrowAnyException();
685+
686+
write1.close();
687+
commit1.close();
688+
commit1Ow.close();
689+
write2.close();
690+
commit2.close();
691+
}
692+
693+
private void commitDvOnly(FileStoreTable table, String user, BinaryRow partition, long commitId)
694+
throws Exception {
695+
// Pick a real data file in the partition; ConflictDetection requires the
696+
// DV to reference an existing data file name.
697+
String dataFileName =
698+
table.store().newScan().withPartitionFilter(Collections.singletonList(partition))
699+
.plan().files().stream()
700+
.findFirst()
701+
.map(e -> e.file().fileName())
702+
.orElseThrow(
703+
() ->
704+
new IllegalStateException(
705+
"No data file in partition for DV commit"));
706+
IndexFileHandler indexFileHandler = table.store().newIndexFileHandler();
707+
BucketedDvMaintainer dvMaintainer =
708+
BucketedDvMaintainer.factory(indexFileHandler)
709+
.create(partition, 0, Collections.emptyList());
710+
dvMaintainer.notifyNewDeletion(dataFileName, 0);
711+
IndexFileMeta dvIndex = dvMaintainer.writeDeletionVectorsIndex().get();
712+
try (TableCommitImpl commit = table.newCommit(user)) {
713+
commit.commit(
714+
commitId,
715+
Collections.singletonList(
716+
new CommitMessageImpl(
717+
partition,
718+
0,
719+
1,
720+
DataIncrement.indexIncrement(
721+
Collections.singletonList(dvIndex)),
722+
CompactIncrement.emptyIncrement())));
723+
}
724+
}
725+
541726
@Test
542727
public void testStrictModeForOverwriteCheckAppend() throws Exception {
543728
String path = tempDir.toString();

0 commit comments

Comments
 (0)