From 49b961d2343f2f2c68c54d1a8b1575fe051f6088 Mon Sep 17 00:00:00 2001 From: tsreaper Date: Wed, 22 Jul 2026 10:21:57 +0800 Subject: [PATCH 1/3] [core] Support overwriting multiple specified partitions --- .../paimon/operation/FileStoreCommit.java | 9 +-- .../paimon/operation/FileStoreCommitImpl.java | 58 ++++++++++++------- .../paimon/table/sink/InnerTableCommit.java | 4 ++ .../paimon/table/sink/TableCommitImpl.java | 24 +++++--- .../java/org/apache/paimon/TestFileStore.java | 3 +- .../paimon/operation/FileDeletionTest.java | 6 +- .../paimon/operation/TestCommitThread.java | 3 +- .../paimon/table/OverwriteTableTest.java | 58 +++++++++++++++++++ 8 files changed, 125 insertions(+), 40 deletions(-) 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..2dd5aa6f588b 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 overwritePartition(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..ec69116d56eb 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 @@ -450,22 +450,45 @@ private boolean containsFileDeletionOrDeletionVectors( return false; } + @Override + public int overwritePartition(Map partition, ManifestCommittable committable) { + if (LOG.isDebugEnabled()) { + LOG.debug( + "Ready to overwrite partition {}\nManifestCommittable: {}", + partition, + committable); + } + return overwritePartition( + () -> { + Predicate partitionPredicate = + createPartitionPredicate( + partition, partitionType, options.partitionDefaultName()); + return PartitionPredicate.fromPredicate(partitionType, partitionPredicate); + }, + committable); + } + @Override public int overwritePartition( - Map partition, - ManifestCommittable committable, - Map properties) { + 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()); - if (LOG.isDebugEnabled()) { - LOG.debug( - "Ready to overwrite partition {}\nManifestCommittable: {}\nProperties: {}", - partition, - committable, - properties); - } 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,22 +527,15 @@ 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"); + + " do not belong to the specified overwrite partitions"); } } } 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..8d75a7334091 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,10 +18,12 @@ 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. */ @@ -30,6 +32,8 @@ 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(List staticPartitions); + /** * If this is set to true, when there is no new data, no snapshot will be generated. By default, * empty commit is ignored. 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..2bfe78ad0a5a 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 overwritePartitions = 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 || overwritePartitions != null) { return true; } return tagAutoManager != null @@ -145,7 +146,15 @@ public boolean forceCreatingSnapshot() { @Override public TableCommitImpl withOverwrite(@Nullable Map overwritePartitions) { - this.overwritePartition = overwritePartitions; + this.overwritePartitionSpec = overwritePartitions; + this.overwritePartitions = null; + return this; + } + + @Override + public TableCommitImpl withOverwrite(List overwritePartitions) { + this.overwritePartitionSpec = null; + this.overwritePartitions = 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 && overwritePartitions == null) { int newSnapshots = 0; for (ManifestCommittable committable : committables) { newSnapshots += commit.commit(committable, checkAppendFiles); @@ -293,8 +302,9 @@ public void commitMultiple(List committables, boolean check committable = new ManifestCommittable(Long.MAX_VALUE); } int newSnapshots = - commit.overwritePartition( - overwritePartition, committable, Collections.emptyMap()); + overwritePartitions == null + ? commit.overwritePartition(overwritePartitionSpec, committable) + : commit.overwritePartition(overwritePartitions, 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..9262330da59b 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.withOverwrite(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.withOverwrite(overwritePartitions).commit(3, messages); + assertThat(read(table)) + .extracting(rowToString) + .containsExactly("7, 2, A, list-again-2A"); + } + } + private void innerTestOverwrite( boolean withPrimaryKey, boolean dynamicPartitionOverwrite, From 3db5e62bceaa1958017b4325180b403ac10ea0a5 Mon Sep 17 00:00:00 2001 From: tsreaper Date: Wed, 22 Jul 2026 18:39:07 +0800 Subject: [PATCH 2/3] Fix comments --- .../paimon/operation/FileStoreCommit.java | 2 +- .../paimon/operation/FileStoreCommitImpl.java | 2 +- .../paimon/table/sink/InnerTableCommit.java | 8 +++++-- .../paimon/table/sink/TableCommitImpl.java | 21 ++++++++++--------- .../paimon/table/OverwriteTableTest.java | 4 ++-- 5 files changed, 21 insertions(+), 16 deletions(-) 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 2dd5aa6f588b..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 @@ -65,7 +65,7 @@ public interface FileStoreCommit extends AutoCloseable { int overwritePartition(Map partition, ManifestCommittable committable); /** Overwrite from manifest committable and specified partitions. */ - int overwritePartition(List partitions, ManifestCommittable committable); + 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 ec69116d56eb..444a34abc7b3 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 @@ -469,7 +469,7 @@ public int overwritePartition(Map partition, ManifestCommittable } @Override - public int overwritePartition( + public int overwriteStaticPartitions( List staticPartitions, ManifestCommittable committable) { checkArgument(!staticPartitions.isEmpty(), "Partitions list cannot be empty."); if (LOG.isDebugEnabled()) { 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 8d75a7334091..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 @@ -30,9 +30,13 @@ 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); - InnerTableCommit withOverwrite(List staticPartitions); + /** + * 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 2bfe78ad0a5a..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 @@ -92,7 +92,7 @@ public class TableCommitImpl implements InnerTableCommit { private final ThreadPoolExecutor fileCheckExecutor; @Nullable private Map overwritePartitionSpec = null; - @Nullable private List overwritePartitions = null; + @Nullable private List overwriteStaticPartitions = null; private boolean batchCommitted = false; private boolean expireForEmptyCommit = true; @@ -136,7 +136,7 @@ public boolean forceCreatingSnapshot() { if (this.forceCreatingSnapshot) { return true; } - if (overwritePartitionSpec != null || overwritePartitions != null) { + if (overwritePartitionSpec != null || overwriteStaticPartitions != null) { return true; } return tagAutoManager != null @@ -145,16 +145,16 @@ public boolean forceCreatingSnapshot() { } @Override - public TableCommitImpl withOverwrite(@Nullable Map overwritePartitions) { - this.overwritePartitionSpec = overwritePartitions; - this.overwritePartitions = null; + public TableCommitImpl withOverwrite(@Nullable Map spec) { + this.overwritePartitionSpec = spec; + this.overwriteStaticPartitions = null; return this; } @Override - public TableCommitImpl withOverwrite(List overwritePartitions) { + public TableCommitImpl withOverwriteStaticPartitions(List overwritePartitions) { this.overwritePartitionSpec = null; - this.overwritePartitions = overwritePartitions; + this.overwriteStaticPartitions = overwritePartitions; return this; } @@ -276,7 +276,7 @@ public void commit(ManifestCommittable committable) { } public void commitMultiple(List committables, boolean checkAppendFiles) { - if (overwritePartitionSpec == null && overwritePartitions == null) { + if (overwritePartitionSpec == null && overwriteStaticPartitions == null) { int newSnapshots = 0; for (ManifestCommittable committable : committables) { newSnapshots += commit.commit(committable, checkAppendFiles); @@ -302,9 +302,10 @@ public void commitMultiple(List committables, boolean check committable = new ManifestCommittable(Long.MAX_VALUE); } int newSnapshots = - overwritePartitions == null + overwriteStaticPartitions == null ? commit.overwritePartition(overwritePartitionSpec, committable) - : commit.overwritePartition(overwritePartitions, committable); + : commit.overwriteStaticPartitions( + overwriteStaticPartitions, committable); maintain( committable.identifier(), maintainExecutor, 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 9262330da59b..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 @@ -109,7 +109,7 @@ public void testOverwriteMultiplePartitions() throws Exception { .map(message -> message.partition().copy()) .distinct() .collect(Collectors.toList()); - commit.withOverwrite(overwritePartitions).commit(1, messages); + commit.withOverwriteStaticPartitions(overwritePartitions).commit(1, messages); assertThat(read(table)) .extracting(rowToString) .containsExactlyInAnyOrder( @@ -124,7 +124,7 @@ public void testOverwriteMultiplePartitions() throws Exception { write.write(overwriteRow(7, 2, "A", "list-again-2A")); messages = write.prepareCommit(true, 3); - commit.withOverwrite(overwritePartitions).commit(3, messages); + commit.withOverwriteStaticPartitions(overwritePartitions).commit(3, messages); assertThat(read(table)) .extracting(rowToString) .containsExactly("7, 2, A, list-again-2A"); From c05444cac32c5c74dad5d4324e239bdad0bee125 Mon Sep 17 00:00:00 2001 From: tsreaper Date: Thu, 23 Jul 2026 10:25:16 +0800 Subject: [PATCH 3/3] Fix test failure --- .../java/org/apache/paimon/operation/FileStoreCommitImpl.java | 2 +- 1 file changed, 1 insertion(+), 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 444a34abc7b3..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 @@ -535,7 +535,7 @@ private int overwritePartition( throw new IllegalArgumentException( "The changes in " + pathFactory.getPartitionString(entry.partition()) - + " do not belong to the specified overwrite partitions"); + + " does not belong to this partition"); } } }