Skip to content

Commit bcd9e9c

Browse files
ZouxxyyZouxxyy
committed
1
1 parent c0f1ffb commit bcd9e9c

File tree

4 files changed

+115
-41
lines changed

4 files changed

+115
-41
lines changed

‎paimon-api/src/main/java/org/apache/paimon/Snapshot.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -496,13 +496,13 @@ public boolean equals(Object o) {
496496
/** Type of changes in this snapshot. */
497497
public enum CommitKind {
498498

499-
/** Changes flushed from the mem table. */
499+
/** New data is appended to the table and no data is removed or deleted. */
500500
APPEND,
501501

502502
/** Changes by compacting existing data files. */
503503
COMPACT,
504504

505-
/** Changes that clear up the whole partition and then add new records. */
505+
/** New data is added to overwrite existing data or just delete existing data. */
506506
OVERWRITE,
507507

508508
/** Collect statistics. */

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

Lines changed: 55 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,6 @@
8787

8888
import static java.util.Collections.emptyList;
8989
import static org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
90-
import static org.apache.paimon.index.HashIndexFile.HASH_INDEX;
9190
import static org.apache.paimon.manifest.ManifestEntry.recordCount;
9291
import static org.apache.paimon.manifest.ManifestEntry.recordCountAdd;
9392
import static org.apache.paimon.manifest.ManifestEntry.recordCountDelete;
@@ -297,22 +296,22 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
297296
List<ManifestEntry> appendChangelog = new ArrayList<>();
298297
List<ManifestEntry> compactTableFiles = new ArrayList<>();
299298
List<ManifestEntry> compactChangelog = new ArrayList<>();
300-
List<IndexManifestEntry> appendHashIndexFiles = new ArrayList<>();
299+
List<IndexManifestEntry> appendIndexFiles = new ArrayList<>();
301300
List<IndexManifestEntry> compactDvIndexFiles = new ArrayList<>();
302301
collectChanges(
303302
committable.fileCommittables(),
304303
appendTableFiles,
305304
appendChangelog,
306305
compactTableFiles,
307306
compactChangelog,
308-
appendHashIndexFiles,
307+
appendIndexFiles,
309308
compactDvIndexFiles);
310309
try {
311310
List<SimpleFileEntry> appendSimpleEntries = SimpleFileEntry.from(appendTableFiles);
312311
if (!ignoreEmptyCommit
313312
|| !appendTableFiles.isEmpty()
314313
|| !appendChangelog.isEmpty()
315-
|| !appendHashIndexFiles.isEmpty()) {
314+
|| !appendIndexFiles.isEmpty()) {
316315
// Optimization for common path.
317316
// Step 1:
318317
// Read manifest entries from changed partitions here and check for conflicts.
@@ -321,6 +320,18 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
321320
// This optimization is mainly used to decrease the number of times we read from
322321
// files.
323322
latestSnapshot = snapshotManager.latestSnapshot();
323+
boolean hasDelete =
324+
appendSimpleEntries.stream()
325+
.anyMatch(entry -> entry.kind().equals(FileKind.DELETE))
326+
|| appendIndexFiles.stream()
327+
.anyMatch(
328+
entry ->
329+
entry.indexFile()
330+
.indexType()
331+
.equals(DELETION_VECTORS_INDEX));
332+
Snapshot.CommitKind commitKind =
333+
hasDelete ? Snapshot.CommitKind.OVERWRITE : Snapshot.CommitKind.APPEND;
334+
324335
if (latestSnapshot != null && checkAppendFiles) {
325336
// it is possible that some partitions only have compact changes,
326337
// so we need to contain all changes
@@ -331,20 +342,20 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
331342
latestSnapshot.commitUser(),
332343
baseEntries,
333344
appendSimpleEntries,
334-
Snapshot.CommitKind.APPEND);
345+
commitKind);
335346
safeLatestSnapshotId = latestSnapshot.id();
336347
}
337348

338349
attempts +=
339350
tryCommit(
340351
appendTableFiles,
341352
appendChangelog,
342-
appendHashIndexFiles,
353+
appendIndexFiles,
343354
committable.identifier(),
344355
committable.watermark(),
345356
committable.logOffsets(),
346357
committable.properties(),
347-
Snapshot.CommitKind.APPEND,
358+
commitKind,
348359
noConflictCheck(),
349360
null);
350361
generatedSnapshot += 1;
@@ -446,15 +457,15 @@ public int overwrite(
446457
List<ManifestEntry> appendChangelog = new ArrayList<>();
447458
List<ManifestEntry> compactTableFiles = new ArrayList<>();
448459
List<ManifestEntry> compactChangelog = new ArrayList<>();
449-
List<IndexManifestEntry> appendHashIndexFiles = new ArrayList<>();
460+
List<IndexManifestEntry> appendIndexFiles = new ArrayList<>();
450461
List<IndexManifestEntry> compactDvIndexFiles = new ArrayList<>();
451462
collectChanges(
452463
committable.fileCommittables(),
453464
appendTableFiles,
454465
appendChangelog,
455466
compactTableFiles,
456467
compactChangelog,
457-
appendHashIndexFiles,
468+
appendIndexFiles,
458469
compactDvIndexFiles);
459470

460471
if (!appendChangelog.isEmpty() || !compactChangelog.isEmpty()) {
@@ -515,7 +526,7 @@ public int overwrite(
515526
tryOverwrite(
516527
partitionFilter,
517528
appendTableFiles,
518-
appendHashIndexFiles,
529+
appendIndexFiles,
519530
committable.identifier(),
520531
committable.watermark(),
521532
committable.logOffsets(),
@@ -658,16 +669,26 @@ public FileIO fileIO() {
658669
return fileIO;
659670
}
660671

672+
/**
673+
* @param appendIndexFiles Index file changes which include hash index files or deletion vectors
674+
* that are not generated by compaction, such as non-pk dv table with row level
675+
* modification.
676+
* @param compactDvIndexFiles Deletion vectors that are generated by compaction, such as pk dv
677+
* table.
678+
*/
661679
private void collectChanges(
662680
List<CommitMessage> commitMessages,
663681
List<ManifestEntry> appendTableFiles,
664682
List<ManifestEntry> appendChangelog,
665683
List<ManifestEntry> compactTableFiles,
666684
List<ManifestEntry> compactChangelog,
667-
List<IndexManifestEntry> appendHashIndexFiles,
685+
List<IndexManifestEntry> appendIndexFiles,
668686
List<IndexManifestEntry> compactDvIndexFiles) {
669687
for (CommitMessage message : commitMessages) {
670688
CommitMessageImpl commitMessage = (CommitMessageImpl) message;
689+
boolean isCompact =
690+
!commitMessage.compactIncrement().compactBefore().isEmpty()
691+
|| !commitMessage.compactIncrement().compactAfter().isEmpty();
671692
commitMessage
672693
.newFilesIncrement()
673694
.newFiles()
@@ -703,44 +724,41 @@ private void collectChanges(
703724
.newIndexFiles()
704725
.forEach(
705726
f -> {
706-
switch (f.indexType()) {
707-
case HASH_INDEX:
708-
appendHashIndexFiles.add(
709-
new IndexManifestEntry(
710-
FileKind.ADD,
711-
commitMessage.partition(),
712-
commitMessage.bucket(),
713-
f));
714-
break;
715-
case DELETION_VECTORS_INDEX:
716-
compactDvIndexFiles.add(
717-
new IndexManifestEntry(
718-
FileKind.ADD,
719-
commitMessage.partition(),
720-
commitMessage.bucket(),
721-
f));
722-
break;
723-
default:
724-
throw new RuntimeException(
725-
"Unknown index type: " + f.indexType());
727+
if (isCompact && f.indexType().equals(DELETION_VECTORS_INDEX)) {
728+
compactDvIndexFiles.add(
729+
new IndexManifestEntry(
730+
FileKind.ADD,
731+
commitMessage.partition(),
732+
commitMessage.bucket(),
733+
f));
734+
} else {
735+
appendIndexFiles.add(
736+
new IndexManifestEntry(
737+
FileKind.ADD,
738+
commitMessage.partition(),
739+
commitMessage.bucket(),
740+
f));
726741
}
727742
});
728743
commitMessage
729744
.indexIncrement()
730745
.deletedIndexFiles()
731746
.forEach(
732747
f -> {
733-
if (f.indexType().equals(DELETION_VECTORS_INDEX)) {
748+
if (isCompact && f.indexType().equals(DELETION_VECTORS_INDEX)) {
734749
compactDvIndexFiles.add(
735750
new IndexManifestEntry(
736751
FileKind.DELETE,
737752
commitMessage.partition(),
738753
commitMessage.bucket(),
739754
f));
740755
} else {
741-
throw new RuntimeException(
742-
"This index type is not supported to delete: "
743-
+ f.indexType());
756+
appendIndexFiles.add(
757+
new IndexManifestEntry(
758+
FileKind.DELETE,
759+
commitMessage.partition(),
760+
commitMessage.bucket(),
761+
f));
744762
}
745763
});
746764
}
@@ -758,8 +776,8 @@ private void collectChanges(
758776
if (!compactChangelog.isEmpty()) {
759777
msg.add(compactChangelog.size() + " compact Changelogs");
760778
}
761-
if (!appendHashIndexFiles.isEmpty()) {
762-
msg.add(appendHashIndexFiles.size() + " append hash index files");
779+
if (!appendIndexFiles.isEmpty()) {
780+
msg.add(appendIndexFiles.size() + " append index files");
763781
}
764782
if (!compactDvIndexFiles.isEmpty()) {
765783
msg.add(compactDvIndexFiles.size() + " compact dv index files");

‎paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818

1919
package org.apache.paimon.spark.sql
2020

21-
import org.apache.paimon.CoreOptions
21+
import org.apache.paimon.{CoreOptions, Snapshot}
2222
import org.apache.paimon.CoreOptions.MergeEngine
2323
import org.apache.paimon.spark.PaimonSparkTestBase
2424

@@ -435,4 +435,32 @@ abstract class DeleteFromTableTestBase extends PaimonSparkTestBase {
435435
}
436436
}
437437
}
438+
439+
test("Paimon delete: non pk table commit kind") {
440+
for (dvEnabled <- Seq(true, false)) {
441+
withTable("t") {
442+
sql(
443+
s"CREATE TABLE t (id INT, data INT) TBLPROPERTIES ('deletion-vectors.enabled' = '$dvEnabled')")
444+
sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS data FROM range(1, 4)")
445+
sql("DELETE FROM t WHERE id = 1")
446+
val table = loadTable("t")
447+
val latestSnapshot = table.latestSnapshot().get()
448+
assert(latestSnapshot.id == 2)
449+
assert(latestSnapshot.commitKind.equals(Snapshot.CommitKind.OVERWRITE))
450+
}
451+
}
452+
}
453+
454+
test("Paimon delete: pk dv table commit kind") {
455+
withTable("t") {
456+
sql(
457+
s"CREATE TABLE t (id INT, data INT) TBLPROPERTIES ('deletion-vectors.enabled' = 'true', 'primary-key' = 'id')")
458+
sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS data FROM range(1, 4)")
459+
sql("DELETE FROM t WHERE id = 1")
460+
val table = loadTable("t")
461+
val latestSnapshot = table.latestSnapshot().get()
462+
assert(latestSnapshot.id == 4)
463+
assert(latestSnapshot.commitKind.equals(Snapshot.CommitKind.COMPACT))
464+
}
465+
}
438466
}

‎paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/UpdateTableTestBase.scala

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818

1919
package org.apache.paimon.spark.sql
2020

21-
import org.apache.paimon.CoreOptions
21+
import org.apache.paimon.{CoreOptions, Snapshot}
2222
import org.apache.paimon.spark.PaimonSparkTestBase
2323
import org.apache.paimon.spark.catalyst.analysis.Update
2424

@@ -357,4 +357,32 @@ abstract class UpdateTableTestBase extends PaimonSparkTestBase {
357357
sql("UPDATE T SET c = 'b' WHERE id = 1")
358358
checkAnswer(sql("SELECT * FROM T"), Seq(Row(1, "s", "b")))
359359
}
360+
361+
test("Paimon update: non pk table commit kind") {
362+
for (dvEnabled <- Seq(true, false)) {
363+
withTable("t") {
364+
sql(
365+
s"CREATE TABLE t (id INT, data INT) TBLPROPERTIES ('deletion-vectors.enabled' = '$dvEnabled')")
366+
sql("INSERT INTO t VALUES (1, 1)")
367+
sql("UPDATE t SET data = 111 WHERE id = 1")
368+
val table = loadTable("t")
369+
val latestSnapshot = table.latestSnapshot().get()
370+
assert(latestSnapshot.id == 2)
371+
assert(latestSnapshot.commitKind.equals(Snapshot.CommitKind.OVERWRITE))
372+
}
373+
}
374+
}
375+
376+
test("Paimon update: pk dv table commit kind") {
377+
withTable("t") {
378+
sql(
379+
s"CREATE TABLE t (id INT, data INT) TBLPROPERTIES ('deletion-vectors.enabled' = 'true', 'primary-key' = 'id')")
380+
sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS data FROM range(1, 4)")
381+
sql("UPDATE t SET data = 111 WHERE id = 1")
382+
val table = loadTable("t")
383+
val latestSnapshot = table.latestSnapshot().get()
384+
assert(latestSnapshot.id == 4)
385+
assert(latestSnapshot.commitKind.equals(Snapshot.CommitKind.COMPACT))
386+
}
387+
}
360388
}

0 commit comments

Comments
 (0)