You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by cz...@apache.org on 2024/12/09 08:41:43 UTC
(paimon) branch master updated: [core] Optimize drop partitions to avoid stack overflow (#4663)
This is an automated email from the ASF dual-hosted git repository.
czweng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 55e4e74012 [core] Optimize drop partitions to avoid stack overflow (#4663)
55e4e74012 is described below
commit 55e4e740128a10a9352be0af7906fb6c7d3bcda9
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Dec 9 16:41:38 2024 +0800
[core] Optimize drop partitions to avoid stack overflow (#4663)
This closes #4663.
---
.../paimon/utils/InternalRowPartitionComputer.java | 18 ++++++++++++
.../paimon/operation/FileStoreCommitImpl.java | 32 ++++++++++++++--------
.../paimon/partition/PartitionPredicate.java | 15 ++++++++++
3 files changed, 54 insertions(+), 11 deletions(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowPartitionComputer.java b/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowPartitionComputer.java
index 8de2720a26..6bb26d7613 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowPartitionComputer.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowPartitionComputer.java
@@ -21,6 +21,7 @@ package org.apache.paimon.utils;
import org.apache.paimon.casting.CastExecutor;
import org.apache.paimon.casting.CastExecutors;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.InternalRow.FieldGetter;
import org.apache.paimon.types.DataType;
@@ -32,6 +33,7 @@ import java.util.List;
import java.util.Map;
import static org.apache.paimon.utils.InternalRowUtils.createNullCheckingFieldGetter;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.TypeUtils.castFromString;
/** PartitionComputer for {@link InternalRow}. */
@@ -102,6 +104,22 @@ public class InternalRowPartitionComputer {
return partValues;
}
+ public static GenericRow convertSpecToInternalRow(
+ Map<String, String> spec, RowType partType, String defaultPartValue) {
+ checkArgument(spec.size() == partType.getFieldCount());
+ GenericRow partRow = new GenericRow(spec.size());
+ List<String> fieldNames = partType.getFieldNames();
+ for (Map.Entry<String, String> entry : spec.entrySet()) {
+ Object value =
+ defaultPartValue.equals(entry.getValue())
+ ? null
+ : castFromString(
+ entry.getValue(), partType.getField(entry.getKey()).type());
+ partRow.setField(fieldNames.indexOf(entry.getKey()), value);
+ }
+ return partRow;
+ }
+
public static String partToSimpleString(
RowType partitionType, BinaryRow partition, String delimiter, int maxLength) {
FieldGetter[] getters = partitionType.fieldGetters();
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 43faadc4d8..547c6e29be 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
@@ -83,6 +83,7 @@ import static org.apache.paimon.index.HashIndexFile.HASH_INDEX;
import static org.apache.paimon.manifest.ManifestEntry.recordCount;
import static org.apache.paimon.manifest.ManifestEntry.recordCountAdd;
import static org.apache.paimon.manifest.ManifestEntry.recordCountDelete;
+import static org.apache.paimon.partition.PartitionPredicate.createBinaryPartitions;
import static org.apache.paimon.partition.PartitionPredicate.createPartitionPredicate;
import static org.apache.paimon.utils.BranchManager.DEFAULT_MAIN_BRANCH;
import static org.apache.paimon.utils.InternalRowPartitionComputer.partToSimpleString;
@@ -530,17 +531,26 @@ public class FileStoreCommitImpl implements FileStoreCommit {
partitions.stream().map(Objects::toString).collect(Collectors.joining(",")));
}
- // partitions may be partial partition fields, so here must to use predicate way.
- Predicate predicate =
- partitions.stream()
- .map(
- partition ->
- createPartitionPredicate(
- partition, partitionType, partitionDefaultName))
- .reduce(PredicateBuilder::or)
- .orElseThrow(() -> new RuntimeException("Failed to get partition filter."));
- PartitionPredicate partitionFilter =
- PartitionPredicate.fromPredicate(partitionType, predicate);
+ boolean fullMode =
+ partitions.stream().allMatch(part -> part.size() == partitionType.getFieldCount());
+ PartitionPredicate partitionFilter;
+ if (fullMode) {
+ List<BinaryRow> binaryPartitions =
+ createBinaryPartitions(partitions, partitionType, partitionDefaultName);
+ partitionFilter = PartitionPredicate.fromMultiple(partitionType, binaryPartitions);
+ } else {
+ // partitions may be partial partition fields, so here must to use predicate way.
+ Predicate predicate =
+ partitions.stream()
+ .map(
+ partition ->
+ createPartitionPredicate(
+ partition, partitionType, partitionDefaultName))
+ .reduce(PredicateBuilder::or)
+ .orElseThrow(
+ () -> new RuntimeException("Failed to get partition filter."));
+ partitionFilter = PartitionPredicate.fromPredicate(partitionType, predicate);
+ }
tryOverwrite(
partitionFilter,
diff --git a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java
index 12ea884be1..1f6c2cfe45 100644
--- a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java
+++ b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java
@@ -19,8 +19,10 @@
package org.apache.paimon.partition;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.data.serializer.InternalSerializers;
import org.apache.paimon.data.serializer.Serializer;
import org.apache.paimon.format.SimpleColStats;
@@ -33,6 +35,7 @@ import org.apache.paimon.utils.RowDataToObjectArrayConverter;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -40,6 +43,7 @@ import java.util.Map;
import java.util.Set;
import static org.apache.paimon.utils.InternalRowPartitionComputer.convertSpecToInternal;
+import static org.apache.paimon.utils.InternalRowPartitionComputer.convertSpecToInternalRow;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -231,4 +235,15 @@ public interface PartitionPredicate {
.map(p -> createPartitionPredicate(p, rowType, defaultPartValue))
.toArray(Predicate[]::new));
}
+
+ static List<BinaryRow> createBinaryPartitions(
+ List<Map<String, String>> partitions, RowType partitionType, String defaultPartValue) {
+ InternalRowSerializer serializer = new InternalRowSerializer(partitionType);
+ List<BinaryRow> result = new ArrayList<>();
+ for (Map<String, String> spec : partitions) {
+ GenericRow row = convertSpecToInternalRow(spec, partitionType, defaultPartValue);
+ result.add(serializer.toBinaryRow(row).copy());
+ }
+ return result;
+ }
}