diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java index 42211d56ca39..a9936833a9de 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java @@ -19,6 +19,7 @@ package org.apache.paimon.operation; import org.apache.paimon.Snapshot; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.disk.IOManager; import org.apache.paimon.fs.FileIO; import org.apache.paimon.manifest.ManifestCommittable; @@ -61,10 +62,10 @@ public interface FileStoreCommit extends AutoCloseable { * note that this partition does not necessarily equal to the partitions of the newly added * key-values. This is just the partition to be cleaned up. */ - int overwritePartition( - Map partition, - ManifestCommittable committable, - Map properties); + int overwritePartition(Map partition, ManifestCommittable committable); + + /** Overwrite from manifest committable and specified partitions. */ + int overwriteStaticPartitions(List partitions, ManifestCommittable committable); /** * Drop multiple partitions. The {@link Snapshot.CommitKind} of generated snapshot is {@link 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 b44032021254..1eacfa71f78d 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 @@ -451,21 +451,44 @@ private boolean containsFileDeletionOrDeletionVectors( } @Override - public int overwritePartition( - Map partition, - ManifestCommittable committable, - Map properties) { - LOG.info( - "Ready to overwrite to table {}, number of commit messages: {}", - tableName, - committable.fileCommittables().size()); + public int overwritePartition(Map partition, ManifestCommittable committable) { if (LOG.isDebugEnabled()) { LOG.debug( - "Ready to overwrite partition {}\nManifestCommittable: {}\nProperties: {}", + "Ready to overwrite partition {}\nManifestCommittable: {}", partition, - committable, - properties); + committable); } + return overwritePartition( + () -> { + Predicate partitionPredicate = + createPartitionPredicate( + partition, partitionType, options.partitionDefaultName()); + return PartitionPredicate.fromPredicate(partitionType, partitionPredicate); + }, + committable); + } + + @Override + public int overwriteStaticPartitions( + List staticPartitions, ManifestCommittable committable) { + checkArgument(!staticPartitions.isEmpty(), "Partitions list cannot be empty."); + if (LOG.isDebugEnabled()) { + LOG.debug( + "Ready to overwrite partitions {}\nManifestCommittable: {}", + staticPartitions, + committable); + } + return overwritePartition( + () -> PartitionPredicate.fromMultiple(partitionType, staticPartitions), + committable); + } + + private int overwritePartition( + Supplier staticPartitionFilter, ManifestCommittable committable) { + LOG.info( + "Ready to overwrite to table {}, number of commit messages: {}", + tableName, + committable.fileCommittables().size()); long started = System.nanoTime(); int generatedSnapshot = 0; @@ -490,7 +513,7 @@ public int overwritePartition( try { boolean skipOverwrite = false; - // partition filter is built from static or dynamic partition according to properties + // partition filter is built from static or dynamic partitions PartitionPredicate partitionFilter = null; if (partitionType.getFieldCount() > 0 && options.dynamicPartitionOverwrite()) { if (changes.appendTableFiles.isEmpty()) { @@ -504,20 +527,13 @@ public int overwritePartition( partitionFilter = PartitionPredicate.fromMultiple(partitionType, partitions); } } else { - // partition may be partial partition fields, so here must use predicate way. - Predicate partitionPredicate = - createPartitionPredicate( - partition, partitionType, options.partitionDefaultName()); - partitionFilter = - PartitionPredicate.fromPredicate(partitionType, partitionPredicate); + partitionFilter = staticPartitionFilter.get(); // sanity check, all changes must be done within the given partition if (partitionFilter != null) { for (ManifestEntry entry : changes.appendTableFiles) { if (!partitionFilter.test(entry.partition())) { throw new IllegalArgumentException( - "Trying to overwrite partition " - + partition - + ", but the changes in " + "The changes in " + pathFactory.getPartitionString(entry.partition()) + " does not belong to this partition"); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java index a73771f2185d..f5bb81d66706 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java @@ -18,17 +18,25 @@ package org.apache.paimon.table.sink; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.metrics.MetricRegistry; import javax.annotation.Nullable; +import java.util.List; import java.util.Map; /** Inner {@link TableCommit} contains overwrite setter. */ public interface InnerTableCommit extends StreamTableCommit, BatchTableCommit { /** Overwrite writing, same as the 'INSERT OVERWRITE T PARTITION (...)' semantics of SQL. */ - InnerTableCommit withOverwrite(@Nullable Map staticPartition); + InnerTableCommit withOverwrite(@Nullable Map spec); + + /** + * Overwrite specified partitions. Unlike {@link InnerTableCommit#withOverwrite}, the given + * partitions must have all the partition keys. + */ + InnerTableCommit withOverwriteStaticPartitions(List staticPartitions); /** * If this is set to true, when there is no new data, no snapshot will be generated. By default, diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java index f5b8c7892bf8..b66cca67e329 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java @@ -21,6 +21,7 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.consumer.ConsumerManager; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.disk.IOManager; import org.apache.paimon.fs.Path; import org.apache.paimon.index.IndexPathFactory; @@ -55,7 +56,6 @@ import java.time.Duration; import java.time.LocalDateTime; import java.util.ArrayList; -import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Map; @@ -91,7 +91,8 @@ public class TableCommitImpl implements InnerTableCommit { private final boolean forceCreatingSnapshot; private final ThreadPoolExecutor fileCheckExecutor; - @Nullable private Map overwritePartition = null; + @Nullable private Map overwritePartitionSpec = null; + @Nullable private List overwriteStaticPartitions = null; private boolean batchCommitted = false; private boolean expireForEmptyCommit = true; @@ -135,7 +136,7 @@ public boolean forceCreatingSnapshot() { if (this.forceCreatingSnapshot) { return true; } - if (overwritePartition != null) { + if (overwritePartitionSpec != null || overwriteStaticPartitions != null) { return true; } return tagAutoManager != null @@ -144,8 +145,16 @@ public boolean forceCreatingSnapshot() { } @Override - public TableCommitImpl withOverwrite(@Nullable Map overwritePartitions) { - this.overwritePartition = overwritePartitions; + public TableCommitImpl withOverwrite(@Nullable Map spec) { + this.overwritePartitionSpec = spec; + this.overwriteStaticPartitions = null; + return this; + } + + @Override + public TableCommitImpl withOverwriteStaticPartitions(List overwritePartitions) { + this.overwritePartitionSpec = null; + this.overwriteStaticPartitions = overwritePartitions; return this; } @@ -267,7 +276,7 @@ public void commit(ManifestCommittable committable) { } public void commitMultiple(List committables, boolean checkAppendFiles) { - if (overwritePartition == null) { + if (overwritePartitionSpec == null && overwriteStaticPartitions == null) { int newSnapshots = 0; for (ManifestCommittable committable : committables) { newSnapshots += commit.commit(committable, checkAppendFiles); @@ -293,8 +302,10 @@ public void commitMultiple(List committables, boolean check committable = new ManifestCommittable(Long.MAX_VALUE); } int newSnapshots = - commit.overwritePartition( - overwritePartition, committable, Collections.emptyMap()); + overwriteStaticPartitions == null + ? commit.overwritePartition(overwritePartitionSpec, committable) + : commit.overwriteStaticPartitions( + overwriteStaticPartitions, committable); maintain( committable.identifier(), maintainExecutor, diff --git a/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java b/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java index 09595d91889c..6f4a9aa89ddc 100644 --- a/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java +++ b/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java @@ -248,8 +248,7 @@ public List overwriteData( null, null, Collections.emptyList(), - (commit, committable) -> - commit.overwritePartition(partition, committable, Collections.emptyMap())); + (commit, committable) -> commit.overwritePartition(partition, committable)); } public Snapshot dropPartitions(List> partitions) { diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java index 580662607941..ee4eccd1d545 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java @@ -153,14 +153,12 @@ public void testMultiPartitions(boolean cleanEmptyDirs) throws Exception { FileStoreCommitImpl commit = store.newCommit(); Map partitionSpec = new HashMap<>(); partitionSpec.put("dt", "0401"); - commit.overwritePartition( - partitionSpec, new ManifestCommittable(commitIdentifier++), Collections.emptyMap()); + commit.overwritePartition(partitionSpec, new ManifestCommittable(commitIdentifier++)); // step 3: generate snapshot 3 by cleaning partition dt=0402/hr=10 partitionSpec.put("dt", "0402"); partitionSpec.put("hr", "8"); - commit.overwritePartition( - partitionSpec, new ManifestCommittable(commitIdentifier++), Collections.emptyMap()); + commit.overwritePartition(partitionSpec, new ManifestCommittable(commitIdentifier++)); commit.close(); // step 4: generate snapshot 4 by cleaning dt=0402/hr=12/bucket-0 diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/TestCommitThread.java b/paimon-core/src/test/java/org/apache/paimon/operation/TestCommitThread.java index cdd2d8fba11f..40b8c6f19273 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/TestCommitThread.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/TestCommitThread.java @@ -171,8 +171,7 @@ private void doOverwrite() throws Exception { () -> commit.overwritePartition( TestKeyValueGenerator.toPartitionMap(partition, MULTI_PARTITIONED), - committable, - Collections.emptyMap())); + committable)); } private void doFinalCompact() { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/OverwriteTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/OverwriteTableTest.java index eccadd38b263..2d1271d1658c 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/OverwriteTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/OverwriteTableTest.java @@ -20,17 +20,20 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.DataFormatTestUtil; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.sink.CommitMessage; import org.apache.paimon.table.sink.InnerTableCommit; import org.apache.paimon.table.sink.StreamTableWrite; import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.TableRead; import org.apache.paimon.types.DataTypes; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; @@ -40,6 +43,8 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.function.Function; +import java.util.stream.Collectors; import static org.apache.paimon.CoreOptions.BUCKET; import static org.apache.paimon.table.SimpleTableTestBase.getResult; @@ -73,6 +78,59 @@ public void testOverwritePrimaryKey( true, dynamicPartitionOverwrite, overwriteData, overwritePartition, expected); } + @Test + public void testOverwriteMultiplePartitions() throws Exception { + Schema schema = + Schema.newBuilder() + .column("pk", DataTypes.INT()) + .column("pt0", DataTypes.INT()) + .column("pt1", DataTypes.STRING()) + .column("v", DataTypes.STRING()) + .partitionKeys("pt0", "pt1") + .option(CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key(), "false") + .build(); + catalog.createTable(identifier(), schema, false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + write( + table, + overwriteRow(1, 1, "A", "old-1A"), + overwriteRow(2, 1, "B", "old-1B"), + overwriteRow(3, 2, "A", "old-2A")); + + Function rowToString = + row -> DataFormatTestUtil.toStringNoRowKind(row, table.rowType()); + try (StreamTableWrite write = table.newWrite(commitUser).withIgnorePreviousFiles(true); + InnerTableCommit commit = table.newCommit(commitUser)) { + write.write(overwriteRow(4, 1, "A", "list-1A")); + write.write(overwriteRow(5, 2, "A", "list-2A")); + List messages = write.prepareCommit(true, 1); + List overwritePartitions = + messages.stream() + .map(message -> message.partition().copy()) + .distinct() + .collect(Collectors.toList()); + commit.withOverwriteStaticPartitions(overwritePartitions).commit(1, messages); + assertThat(read(table)) + .extracting(rowToString) + .containsExactlyInAnyOrder( + "2, 1, B, old-1B", "4, 1, A, list-1A", "5, 2, A, list-2A"); + + write.write(overwriteRow(6, 1, "A", "map-1A")); + messages = write.prepareCommit(true, 2); + commit.withOverwrite(Collections.singletonMap("pt0", "1")).commit(2, messages); + assertThat(read(table)) + .extracting(rowToString) + .containsExactlyInAnyOrder("5, 2, A, list-2A", "6, 1, A, map-1A"); + + write.write(overwriteRow(7, 2, "A", "list-again-2A")); + messages = write.prepareCommit(true, 3); + commit.withOverwriteStaticPartitions(overwritePartitions).commit(3, messages); + assertThat(read(table)) + .extracting(rowToString) + .containsExactly("7, 2, A, list-again-2A"); + } + } + private void innerTestOverwrite( boolean withPrimaryKey, boolean dynamicPartitionOverwrite,