diff --git a/docs/docs/primary-key-table/data-distribution.md b/docs/docs/primary-key-table/data-distribution.md
index b7a0fecb7c64..360febf904d9 100644
--- a/docs/docs/primary-key-table/data-distribution.md
+++ b/docs/docs/primary-key-table/data-distribution.md
@@ -70,16 +70,18 @@ Postpone bucket mode is configured by `'bucket' = '-2'`.
This mode aims to solve the difficulty to determine a fixed number of buckets
and support different buckets for different partitions.
-By default, `postpone.batch-write-fixed-bucket` is `true`. This staged fixed-bucket flow uses
-Spark's DataSource V1 write path, even when `spark.paimon.write.use-v2-write` is enabled. Spark
-completes each batch in three steps:
+By default, `postpone.batch-write-fixed-bucket` is `true`. The fixed-bucket flow uses Spark's
+DataSource V1 write path, even when `spark.paimon.write.use-v2-write` is enabled. Unless direct
+writing applies, Spark completes each batch in three steps:
1. Write the current batch to uncommitted bucket `-2` files. Spark derives each partition's row
count and file size directly from the staged file metadata; there is no extra input scan, cache,
or per-row statistics pass.
-2. Calculate the required bucket number per touched partition. `postpone.target-row-num-per-bucket`,
- when configured, takes precedence over `postpone.target-size-per-bucket` (default `1 GB`). The
- result is at least `1`, rounded up to a power of two, and capped by
+2. Calculate the required bucket number per touched partition. For a partition without real
+ buckets, an explicitly configured `postpone.default-bucket-num` is used exactly. Otherwise,
+ `postpone.target-row-num-per-bucket`, when configured, takes precedence over
+ `postpone.target-size-per-bucket` (default `1 GB`). An inferred result is at least `1`, rounded
+ up to a power of two, and capped by
`postpone.batch-write-fixed-bucket.max-parallelism`.
3. Route the staged records to real buckets and commit them. The current batch becomes visible only
in this commit.
@@ -91,6 +93,14 @@ larger than the existing layout. Different partitions may have different target
The rescale is a separate overwrite commit which changes real buckets only; the current batch is
appended in the following commit.
+`postpone.default-bucket-num` has no default value. When it is explicitly configured, Spark can
+skip the staged bucket `-2` files and write directly to real buckets for `INSERT OVERWRITE`, or
+when the base snapshot contains no real buckets. An overwrite always uses the configured number
+exactly and does not rescale the replaced layout. An append to an existing real-bucket partition
+ignores this option and still uses the staged batch to decide whether rescaling is required. If a
+batch mixes existing and new real-bucket partitions, the whole batch remains staged; only the new
+partitions use the configured default.
+
Previously committed bucket `-2` files are not included in the calculation, read, rewritten, or
deleted by an append or rescale. They remain available to merge-on-read and regular postpone
compaction. `INSERT OVERWRITE` still follows its normal replacement semantics.
@@ -100,14 +110,13 @@ records are first stored in the `bucket-postpone` directory of each partition
and are not available to readers.
To move these records into the correct bucket and make them readable, run a compaction job.
See `compact` [procedure](../flink/procedures).
-The bucket number for the partitions compacted for the first time
-is configured by the option `postpone.default-bucket-num`, whose default value is `1`.
-You can also configure `postpone.target-row-num-per-bucket` to calculate the bucket number
-from the row count of the files in the postpone bucket directory.
-The calculated bucket number is `ceil(row_count / postpone.target-row-num-per-bucket)`,
-and is at least `1`.
-When this option is configured, it takes precedence over `postpone.default-bucket-num`
-for partitions compacted for the first time.
+The bucket number for partitions compacted for the first time can be configured by the option
+`postpone.default-bucket-num`. Its value is used exactly and takes precedence over automatic
+estimation. Otherwise, `postpone.target-row-num-per-bucket`, when configured, calculates the
+bucket number as `ceil(row_count / target_row_count)`. If it is not configured, Paimon calculates
+the bucket number as `ceil(postpone_file_size / postpone.target-size-per-bucket)`; the target size
+defaults to `1 GB`. Both estimates are at least `1`. Execution parallelism does not determine the
+logical bucket number.
Partitions that already have real bucket files keep their existing bucket number.
Finally, when you feel that the bucket number of some partition is too small,
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index 0c0c1a0a8d46..a7c8114dd8b4 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1316,9 +1316,9 @@
postpone.default-bucket-num |
- 1 |
+ (none) |
Integer |
- Bucket number for the partitions compacted for the first time in postpone bucket tables. |
+ Optional bucket number for partitions receiving real buckets for the first time and for fixed-bucket overwrite writes. The configured value is used exactly and takes precedence over automatic bucket estimation. When unset, Paimon estimates the bucket number from the target row count or target file size. |
postpone.merge-on-read |
@@ -1330,13 +1330,13 @@
postpone.target-row-num-per-bucket |
(none) |
Long |
- Target row number per bucket when estimating the required bucket number from the current staged batch or compacting postpone bucket files. |
+ Target postpone row count per bucket when estimating the required bucket number from staged or committed postpone files. When configured, this option takes precedence over 'postpone.target-size-per-bucket'. |
postpone.target-size-per-bucket |
1 gb |
MemorySize |
- Target staged file size per bucket when Spark estimates the required bucket number from the current staged batch. This option is ignored when 'postpone.target-row-num-per-bucket' is configured. |
+ Target postpone file size per bucket when estimating the required bucket number from staged or committed postpone files. This option is ignored when 'postpone.target-row-num-per-bucket' is configured. |
primary-key |
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 96fdbf731b0b..3be4d072e121 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2766,23 +2766,23 @@ public String toString() {
public static final ConfigOption POSTPONE_DEFAULT_BUCKET_NUM =
key("postpone.default-bucket-num")
.intType()
- .defaultValue(1)
+ .noDefaultValue()
.withDescription(
- "Bucket number for the partitions compacted for the first time in postpone bucket tables.");
+ "Optional bucket number for partitions receiving real buckets for the first time and for fixed-bucket overwrite writes. The configured value is used exactly and takes precedence over automatic bucket estimation. When unset, Paimon estimates the bucket number from the target row count or target file size.");
public static final ConfigOption POSTPONE_TARGET_ROW_NUM_PER_BUCKET =
key("postpone.target-row-num-per-bucket")
.longType()
.noDefaultValue()
.withDescription(
- "Target row number per bucket when estimating the required bucket number from the current staged batch or compacting postpone bucket files.");
+ "Target postpone row count per bucket when estimating the required bucket number from staged or committed postpone files. When configured, this option takes precedence over 'postpone.target-size-per-bucket'.");
public static final ConfigOption POSTPONE_TARGET_SIZE_PER_BUCKET =
key("postpone.target-size-per-bucket")
.memoryType()
.defaultValue(MemorySize.parse("1 gb"))
.withDescription(
- "Target staged file size per bucket when Spark estimates the required bucket number from the current staged batch. "
+ "Target postpone file size per bucket when estimating the required bucket number from staged or committed postpone files. "
+ "This option is ignored when 'postpone.target-row-num-per-bucket' is configured.");
public static final ConfigOption GLOBAL_INDEX_ROW_COUNT_PER_SHARD =
@@ -4465,8 +4465,15 @@ public int postponeBatchWriteFixedBucketRescaleLoadFactor() {
return options.get(POSTPONE_BATCH_WRITE_FIXED_BUCKET_RESCALE_LOAD_FACTOR);
}
- public int postponeDefaultBucketNum() {
- return options.get(POSTPONE_DEFAULT_BUCKET_NUM);
+ public Optional postponeDefaultBucketNum() {
+ Optional bucketNum = options.getOptional(POSTPONE_DEFAULT_BUCKET_NUM);
+ bucketNum.ifPresent(
+ value ->
+ checkArgument(
+ value > 0,
+ "Option '%s' must be greater than 0.",
+ POSTPONE_DEFAULT_BUCKET_NUM.key()));
+ return bucketNum;
}
public Optional postponeTargetRowNumPerBucket() {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java
index 0d2a77ad0b54..7ea59f78a99f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java
@@ -127,51 +127,59 @@ public static List groupPostponeFiles(List splits) {
return result;
}
- public static PostponeBucketAssigner createPostponeBucketAssigner(
- FileStoreTable table, long snapshotId, int defaultParallelism) {
- return loadPostponeBucketAssigner(table, snapshotId, defaultParallelism, null);
- }
-
- private static PostponeBucketAssigner loadPostponeBucketAssigner(
- FileStoreTable table,
- long snapshotId,
- int defaultParallelism,
- @Nullable PartitionPredicate partitionFilter) {
- Map knownNumBuckets =
- getKnownNumBuckets(table, snapshotId, partitionFilter);
- Map postponeRowCounts =
- !table.coreOptions().postponeTargetRowNumPerBucket().isPresent()
- ? Collections.emptyMap()
- : getPostponeRowCounts(table, snapshotId, partitionFilter);
- Long targetRowNumPerBucket =
- table.coreOptions().postponeTargetRowNumPerBucket().orElse(null);
- int defaultBucketNum =
- table.coreOptions()
- .toConfiguration()
- .contains(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM)
- ? table.coreOptions().postponeDefaultBucketNum()
- : defaultParallelism;
- return new PostponeBucketAssigner(
- knownNumBuckets, targetRowNumPerBucket, postponeRowCounts, defaultBucketNum);
- }
-
- /** Creates snapshot-bound routing metadata. */
+ public static PostponeBucketNumResolver createPostponeBucketNumResolver(
+ FileStoreTable table, long snapshotId) {
+ return loadPostponeBucketNumResolver(table, snapshotId, null);
+ }
+
+ private static PostponeBucketNumResolver loadPostponeBucketNumResolver(
+ FileStoreTable table, long snapshotId, @Nullable List postponePartitions) {
+ CoreOptions options = table.coreOptions();
+ Map numBucketsByPartition =
+ postponePartitions == null
+ ? getKnownNumBuckets(table, snapshotId)
+ : getKnownNumBuckets(table, snapshotId, postponePartitions);
+ Integer configuredDefaultBucketNum = options.postponeDefaultBucketNum().orElse(null);
+ if (configuredDefaultBucketNum == null) {
+ Optional targetRowNumPerBucket = options.postponeTargetRowNumPerBucket();
+ if (targetRowNumPerBucket.isPresent()) {
+ checkArgument(
+ targetRowNumPerBucket.get() > 0,
+ "Option '%s' must be greater than 0.",
+ CoreOptions.POSTPONE_TARGET_ROW_NUM_PER_BUCKET.key());
+ addEstimatedBucketNums(
+ numBucketsByPartition,
+ getPostponeRowCounts(
+ postponeFileIterator(table, snapshotId, postponePartitions)),
+ targetRowNumPerBucket.get(),
+ CoreOptions.POSTPONE_TARGET_ROW_NUM_PER_BUCKET.key());
+ } else {
+ long targetSizePerBucket = options.postponeTargetSizePerBucket();
+ checkArgument(
+ targetSizePerBucket > 0,
+ "Option '%s' must be greater than 0.",
+ CoreOptions.POSTPONE_TARGET_SIZE_PER_BUCKET.key());
+ addEstimatedBucketNums(
+ numBucketsByPartition,
+ getPostponeFileSizes(
+ postponeFileIterator(table, snapshotId, postponePartitions)),
+ targetSizePerBucket,
+ CoreOptions.POSTPONE_TARGET_SIZE_PER_BUCKET.key());
+ }
+ }
+ return new PostponeBucketNumResolver(numBucketsByPartition, configuredDefaultBucketNum);
+ }
+
+ /** Creates snapshot-bound routing metadata for partitions containing postpone files. */
public static PostponeBucketRouter createPostponeBucketRouter(
- FileStoreTable table,
- long snapshotId,
- int defaultParallelism,
- @Nullable PartitionPredicate partitionFilter) {
+ FileStoreTable table, long snapshotId, List postponePartitions) {
return newPostponeBucketRouter(
- table,
- loadPostponeBucketAssigner(table, snapshotId, defaultParallelism, partitionFilter));
+ table, loadPostponeBucketNumResolver(table, snapshotId, postponePartitions));
}
/** Creates routing metadata from bucket numbers decided by an execution engine. */
public static PostponeBucketRouter createPostponeBucketRouter(
- FileStoreTable table,
- Map numBucketsByPartition,
- int defaultBucketNum) {
- checkArgument(defaultBucketNum > 0, "Default postpone bucket number must be positive.");
+ FileStoreTable table, Map numBucketsByPartition) {
Map copied = new HashMap<>();
for (Map.Entry entry : numBucketsByPartition.entrySet()) {
checkArgument(
@@ -179,13 +187,11 @@ public static PostponeBucketRouter createPostponeBucketRouter(
"Postpone bucket number must be positive.");
copied.put(entry.getKey().copy(), entry.getValue());
}
- return newPostponeBucketRouter(
- table,
- new PostponeBucketAssigner(copied, null, Collections.emptyMap(), defaultBucketNum));
+ return newPostponeBucketRouter(table, new PostponeBucketNumResolver(copied, null));
}
private static PostponeBucketRouter newPostponeBucketRouter(
- FileStoreTable table, PostponeBucketAssigner bucketAssigner) {
+ FileStoreTable table, PostponeBucketNumResolver bucketNumResolver) {
List trimmedPrimaryKeys = table.schema().trimmedPrimaryKeys();
int[] bucketKeyMapping =
table.schema().bucketKeys().stream()
@@ -202,7 +208,7 @@ private static PostponeBucketRouter newPostponeBucketRouter(
PrimaryKeyTableUtils.PrimaryKeyFieldsExtractor.EXTRACTOR.keyFields(
table.schema()));
return new PostponeBucketRouter(
- bucketAssigner,
+ bucketNumResolver,
keyType,
table.schema().logicalBucketKeyType(),
bucketKeyMapping,
@@ -210,52 +216,57 @@ private static PostponeBucketRouter newPostponeBucketRouter(
}
public static int computeBucketNumByRowCount(long rowCount, long targetRowNumPerBucket) {
- if (targetRowNumPerBucket <= 0) {
- throw new IllegalArgumentException(
- "Option 'postpone.target-row-num-per-bucket' must be greater than 0.");
- }
+ return computeBucketNum(
+ rowCount,
+ targetRowNumPerBucket,
+ CoreOptions.POSTPONE_TARGET_ROW_NUM_PER_BUCKET.key());
+ }
- long bucketNum = rowCount <= 0 ? 1 : (rowCount - 1) / targetRowNumPerBucket + 1;
+ private static int computeBucketNum(
+ long value, long targetValuePerBucket, String targetOptionKey) {
+ checkArgument(
+ targetValuePerBucket > 0, "Option '%s' must be greater than 0.", targetOptionKey);
+ long bucketNum = value <= 0 ? 1 : (value - 1) / targetValuePerBucket + 1;
if (bucketNum > Integer.MAX_VALUE) {
throw new IllegalArgumentException(
"Computed postpone bucket number "
+ bucketNum
+ " exceeds the maximum integer value (Integer.MAX_VALUE = "
+ Integer.MAX_VALUE
- + "). Consider increasing 'postpone.target-row-num-per-bucket' "
+ + "). Consider increasing '"
+ + targetOptionKey
+ + "' "
+ "to reduce the bucket count.");
}
return (int) bucketNum;
}
- public static int determineBucketNum(
- BinaryRow partition,
- Map knownNumBuckets,
- Optional targetRowNumPerBucket,
- Map postponeRowCounts,
- int defaultBucketNum) {
- return determineBucketNum(
- partition,
- knownNumBuckets,
- targetRowNumPerBucket.orElse(null),
- postponeRowCounts,
- defaultBucketNum);
- }
-
- public static int determineBucketNum(
+ private static void addEstimatedBucketNums(
+ Map numBucketsByPartition,
+ Map valuesByPartition,
+ long targetValuePerBucket,
+ String targetOptionKey) {
+ for (Map.Entry entry : valuesByPartition.entrySet()) {
+ if (!numBucketsByPartition.containsKey(entry.getKey())) {
+ numBucketsByPartition.put(
+ entry.getKey(),
+ computeBucketNum(entry.getValue(), targetValuePerBucket, targetOptionKey));
+ }
+ }
+ }
+
+ static int determineBucketNum(
BinaryRow partition,
- Map knownNumBuckets,
- @Nullable Long targetRowNumPerBucket,
- Map postponeRowCounts,
- int defaultBucketNum) {
- Integer knownBucketNum = knownNumBuckets.get(partition);
- if (knownBucketNum != null) {
- return knownBucketNum;
- } else if (targetRowNumPerBucket != null) {
- return computeBucketNumByRowCount(
- postponeRowCounts.getOrDefault(partition, 0L), targetRowNumPerBucket);
+ Map numBucketsByPartition,
+ @Nullable Integer configuredDefaultBucketNum) {
+ Integer numBuckets = numBucketsByPartition.get(partition);
+ if (numBuckets != null) {
+ return numBuckets;
+ } else if (configuredDefaultBucketNum != null) {
+ return configuredDefaultBucketNum;
} else {
- return defaultBucketNum;
+ throw new IllegalArgumentException(
+ "Missing postpone bucket number for partition " + partition + ".");
}
}
@@ -278,6 +289,11 @@ public static FixedBucketDecision decideFixedBucketNum(
existingBucketNum == null || existingBucketNum > 0,
"Existing bucket number must be positive.");
+ Optional configuredDefaultBucketNum = options.postponeDefaultBucketNum();
+ if (existingBucketNum == null && configuredDefaultBucketNum.isPresent()) {
+ return new FixedBucketDecision(configuredDefaultBucketNum.get(), false);
+ }
+
int maxBucketNum = options.postponeBatchWriteFixedBucketMaxParallelism();
checkArgument(
maxBucketNum > 0,
@@ -453,48 +469,69 @@ private static Map getPostponeRowCounts(Iterator
return rowCounts;
}
- public static FileStoreTable tableForPostponeCompact(
+ private static Iterator postponeFileIterator(
+ FileStoreTable table, long snapshotId, @Nullable List postponePartitions) {
+ SnapshotReader reader =
+ table.newSnapshotReader()
+ .withSnapshot(snapshotId)
+ .withBucket(BucketMode.POSTPONE_BUCKET);
+ if (postponePartitions != null) {
+ reader.withPartitionFilter(postponePartitions);
+ }
+ return reader.readFileIterator();
+ }
+
+ static Map getPostponeFileSizes(
+ FileStoreTable table, long snapshotId, @Nullable PartitionPredicate partitionFilter) {
+ SnapshotReader reader =
+ table.newSnapshotReader()
+ .withSnapshot(snapshotId)
+ .withBucket(BucketMode.POSTPONE_BUCKET);
+ if (partitionFilter != null) {
+ reader.withPartitionFilter(partitionFilter);
+ }
+
+ return getPostponeFileSizes(reader.readFileIterator());
+ }
+
+ private static Map getPostponeFileSizes(Iterator iterator) {
+ Map fileSizes = new HashMap<>();
+ while (iterator.hasNext()) {
+ ManifestEntry entry = iterator.next();
+ fileSizes.merge(
+ entry.partition(),
+ entry.file().fileSize(),
+ (left, right) -> Math.addExact(left, right));
+ }
+ return fileSizes;
+ }
+
+ public static FileStoreTable tableForPostponeRewrite(
FileStoreTable table, int numBuckets, long snapshotId) {
- Map compactOptions = new HashMap<>();
- compactOptions.put(BUCKET.key(), String.valueOf(numBuckets));
- compactOptions.put(WRITE_ONLY.key(), "false");
- compactOptions.put(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), String.valueOf(snapshotId));
- return table.copy(compactOptions);
+ Map rewriteOptions = new HashMap<>();
+ rewriteOptions.put(BUCKET.key(), String.valueOf(numBuckets));
+ rewriteOptions.put(WRITE_ONLY.key(), "false");
+ rewriteOptions.put(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), String.valueOf(snapshotId));
+ return table.copy(rewriteOptions);
}
- /** Snapshot-bound bucket-count assignment. */
- public static final class PostponeBucketAssigner implements Serializable {
+ /** Resolves the snapshot-bound bucket count of a partition. */
+ public static final class PostponeBucketNumResolver implements Serializable {
private static final long serialVersionUID = 1L;
- private final Map knownNumBuckets;
- @Nullable private final Long targetRowNumPerBucket;
- private final Map postponeRowCounts;
- private final int defaultBucketNum;
+ private final Map numBucketsByPartition;
+ @Nullable private final Integer configuredDefaultBucketNum;
- private PostponeBucketAssigner(
- Map knownNumBuckets,
- @Nullable Long targetRowNumPerBucket,
- Map postponeRowCounts,
- int defaultBucketNum) {
- this.knownNumBuckets = knownNumBuckets;
- this.targetRowNumPerBucket = targetRowNumPerBucket;
- this.postponeRowCounts = postponeRowCounts;
- this.defaultBucketNum = defaultBucketNum;
+ private PostponeBucketNumResolver(
+ Map numBucketsByPartition,
+ @Nullable Integer configuredDefaultBucketNum) {
+ this.numBucketsByPartition = numBucketsByPartition;
+ this.configuredDefaultBucketNum = configuredDefaultBucketNum;
}
- public int assign(BinaryRow partition) {
- return determineBucketNum(
- partition,
- knownNumBuckets,
- targetRowNumPerBucket,
- postponeRowCounts,
- defaultBucketNum);
- }
-
- private PostponeBucketAssigner withDefaultBucketNum(int newDefaultBucketNum) {
- return new PostponeBucketAssigner(
- knownNumBuckets, targetRowNumPerBucket, postponeRowCounts, newDefaultBucketNum);
+ public int numBuckets(BinaryRow partition) {
+ return determineBucketNum(partition, numBucketsByPartition, configuredDefaultBucketNum);
}
}
@@ -523,7 +560,7 @@ public static final class PostponeBucketRouter implements Serializable {
private static final long serialVersionUID = 1L;
- private final PostponeBucketAssigner bucketAssigner;
+ private final PostponeBucketNumResolver bucketNumResolver;
private final RowType keyType;
private final RowType bucketKeyType;
private final int[] bucketKeyMapping;
@@ -532,12 +569,12 @@ public static final class PostponeBucketRouter implements Serializable {
@Nullable private transient BucketFunction bucketFunction;
private PostponeBucketRouter(
- PostponeBucketAssigner bucketAssigner,
+ PostponeBucketNumResolver bucketNumResolver,
RowType keyType,
RowType bucketKeyType,
int[] bucketKeyMapping,
CoreOptions.BucketFunctionType bucketFunctionType) {
- this.bucketAssigner = bucketAssigner;
+ this.bucketNumResolver = bucketNumResolver;
this.keyType = keyType;
this.bucketKeyType = bucketKeyType;
this.bucketKeyMapping = bucketKeyMapping;
@@ -556,18 +593,7 @@ public int bucket(BinaryRow partition, InternalRow trimmedPrimaryKey) {
}
public int numBuckets(BinaryRow partition) {
- return bucketAssigner.assign(partition);
- }
-
- public PostponeBucketRouter withDefaultBucketNum(int newDefaultBucketNum) {
- checkArgument(
- newDefaultBucketNum > 0, "Default postpone bucket number must be positive.");
- return new PostponeBucketRouter(
- bucketAssigner.withDefaultBucketNum(newDefaultBucketNum),
- keyType,
- bucketKeyType,
- bucketKeyMapping,
- bucketFunctionType);
+ return bucketNumResolver.numBuckets(partition);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergePlan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergePlan.java
index a65e415ee07c..b2e6bc460e6f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergePlan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergePlan.java
@@ -92,16 +92,6 @@ public long numPotentialBuckets() {
return numPotentialBuckets;
}
- PostponeMergePlan withDefaultBucketNum(int newDefaultBucketNum) {
- return new PostponeMergePlan(
- realSplits,
- postponeSplits,
- bucketRouter.withDefaultBucketNum(newDefaultBucketNum),
- keyType,
- resultReadType,
- mergeReadType);
- }
-
private static long numPotentialBuckets(
List realSplits,
List postponeSplits,
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeReadBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeReadBuilder.java
index c0561c09db17..147d12e05a69 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeReadBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeReadBuilder.java
@@ -21,6 +21,7 @@
import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValueFileStore;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
@@ -40,6 +41,7 @@
import java.io.Serializable;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.Set;
@@ -63,7 +65,6 @@ public final class PostponeMergeReadBuilder implements Serializable {
@Nullable private RowType readType;
@Nullable private transient MetricRegistry metricRegistry;
@Nullable private transient String readProtectionTagName;
- private int defaultBucketNum = 1;
private PostponeMergeReadBuilder(FileStoreTable table, @Nullable Snapshot snapshot) {
this.table = table;
@@ -180,12 +181,6 @@ public PostponeMergeReadBuilder withMetricRegistry(MetricRegistry metricRegistry
return this;
}
- public PostponeMergeReadBuilder withDefaultBucketNum(int defaultBucketNum) {
- checkArgument(defaultBucketNum > 0, "Default postpone bucket number must be positive.");
- this.defaultBucketNum = defaultBucketNum;
- return this;
- }
-
public PostponeMergePlan plan() {
checkArgument(snapshot != null, "Snapshot-bound postpone merge plan requires a snapshot.");
RowType resultReadType = resultReadType();
@@ -218,12 +213,28 @@ public PostponeMergePlan plan() {
postponeReader.withPartitionFilter(partitionFilter);
}
+ List realSplits = realReader.read().dataSplits();
+ List postponeSplits =
+ PostponeUtils.groupPostponeFiles(postponeReader.read().dataSplits());
+ PostponeUtils.PostponeBucketRouter bucketRouter;
+ if (postponeSplits.isEmpty()) {
+ bucketRouter = PostponeUtils.createPostponeBucketRouter(table, Collections.emptyMap());
+ } else {
+ List postponePartitions =
+ postponeSplits.stream()
+ .map(DataSplit::partition)
+ .distinct()
+ .collect(Collectors.toList());
+ bucketRouter =
+ PostponeUtils.createPostponeBucketRouter(
+ table, snapshot.id(), postponePartitions);
+ }
+
PostponeMergePlan plan =
new PostponeMergePlan(
- realReader.read().dataSplits(),
- PostponeUtils.groupPostponeFiles(postponeReader.read().dataSplits()),
- PostponeUtils.createPostponeBucketRouter(
- table, snapshot.id(), defaultBucketNum, partitionFilter),
+ realSplits,
+ postponeSplits,
+ bucketRouter,
keyType(),
resultReadType,
mergeReadType);
@@ -246,18 +257,6 @@ public PostponeMergePlan plan(
mergeReadType(resultReadType));
}
- /** Rebuilds only the routing metadata of an existing plan with a new default bucket number. */
- public PostponeMergePlan reroute(PostponeMergePlan plan, int newDefaultBucketNum) {
- checkArgument(newDefaultBucketNum > 0, "Default postpone bucket number must be positive.");
- if (table.coreOptions()
- .toConfiguration()
- .contains(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM)) {
- return plan;
- }
- defaultBucketNum = newDefaultBucketNum;
- return plan.withDefaultBucketNum(defaultBucketNum);
- }
-
@Nullable
public String readProtectionTagName() {
return readProtectionTagName;
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java
index 1d07359b6db9..d6c6ddc5d3e2 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java
@@ -117,6 +117,28 @@ public void testGetPostponeRowCountsFromSnapshot() {
verify(reader).withPartitionFilter(partitionFilter);
}
+ @Test
+ public void testGetPostponeFileSizesFromSnapshot() {
+ BinaryRow partition = partition(1);
+ PartitionPredicate partitionFilter = mock(PartitionPredicate.class);
+ DataFileMeta file = mock(DataFileMeta.class);
+ when(file.fileSize()).thenReturn(1024L);
+ ManifestEntry entry = mock(ManifestEntry.class);
+ when(entry.partition()).thenReturn(partition);
+ when(entry.file()).thenReturn(file);
+
+ SnapshotReader reader = mock(SnapshotReader.class, RETURNS_SELF);
+ when(reader.readFileIterator()).thenReturn(Collections.singletonList(entry).iterator());
+ FileStoreTable table = mock(FileStoreTable.class);
+ when(table.newSnapshotReader()).thenReturn(reader);
+
+ assertThat(PostponeUtils.getPostponeFileSizes(table, 5L, partitionFilter))
+ .containsEntry(partition, 1024L);
+ verify(reader).withSnapshot(5L);
+ verify(reader).withBucket(BucketMode.POSTPONE_BUCKET);
+ verify(reader).withPartitionFilter(partitionFilter);
+ }
+
@Test
public void testGetLevel0BucketsFromSnapshot() {
BinaryRow partition = partition(1);
@@ -223,7 +245,7 @@ public void testTableForPostponeCompact() {
FileStoreTable copied = mock(FileStoreTable.class);
when(table.copy(anyMap())).thenReturn(copied);
- assertThat(PostponeUtils.tableForPostponeCompact(table, 4, 5L)).isSameAs(copied);
+ assertThat(PostponeUtils.tableForPostponeRewrite(table, 4, 5L)).isSameAs(copied);
@SuppressWarnings("unchecked")
ArgumentCaptor> options = ArgumentCaptor.forClass(Map.class);
@@ -262,33 +284,24 @@ public void testComputeBucketNumByRowCountRejectsOverflow() {
@Test
public void testDetermineBucketNum() {
- Map knownNumBuckets = new HashMap<>();
- Map postponeRowCounts = new HashMap<>();
+ Map numBucketsByPartition = new HashMap<>();
BinaryRow knownPartition = partition(1);
- BinaryRow targetPartition = partition(2);
- BinaryRow defaultPartition = partition(3);
+ BinaryRow configuredPartition = partition(2);
+ BinaryRow missingPartition = partition(3);
- knownNumBuckets.put(knownPartition, 4);
- postponeRowCounts.put(knownPartition, 1000L);
- postponeRowCounts.put(targetPartition, 450L);
+ numBucketsByPartition.put(knownPartition, 4);
- assertThat(
- PostponeUtils.determineBucketNum(
- knownPartition, knownNumBuckets, 200L, postponeRowCounts, 1))
+ assertThat(PostponeUtils.determineBucketNum(knownPartition, numBucketsByPartition, 7))
.isEqualTo(4);
- assertThat(
- PostponeUtils.determineBucketNum(
- targetPartition, knownNumBuckets, 200L, postponeRowCounts, 1))
- .isEqualTo(3);
- assertThat(
- PostponeUtils.determineBucketNum(
- defaultPartition,
- knownNumBuckets,
- (Long) null,
- postponeRowCounts,
- 7))
+ assertThat(PostponeUtils.determineBucketNum(configuredPartition, numBucketsByPartition, 7))
.isEqualTo(7);
+ assertThatThrownBy(
+ () ->
+ PostponeUtils.determineBucketNum(
+ missingPartition, numBucketsByPartition, null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Missing postpone bucket number");
}
@Test
@@ -341,6 +354,32 @@ public void testDecideFixedBucketNum() {
29, 0, 7, CoreOptions.fromMap(lowerLoadFactorOptions));
assertThat(lowerLoadFactor.targetBucketNum()).isEqualTo(16);
assertThat(lowerLoadFactor.requiresRescale()).isTrue();
+
+ Map configuredDefaultOptions = new HashMap<>(optionMap);
+ configuredDefaultOptions.put(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM.key(), "3");
+ CoreOptions optionsWithDefault = CoreOptions.fromMap(configuredDefaultOptions);
+ PostponeUtils.FixedBucketDecision configuredDefault =
+ PostponeUtils.decideFixedBucketNum(100, 0, null, optionsWithDefault);
+ assertThat(configuredDefault.targetBucketNum()).isEqualTo(3);
+ assertThat(configuredDefault.requiresRescale()).isFalse();
+
+ PostponeUtils.FixedBucketDecision existingIgnoresDefault =
+ PostponeUtils.decideFixedBucketNum(225, 0, 7, optionsWithDefault);
+ assertThat(existingIgnoresDefault.targetBucketNum()).isEqualTo(16);
+ assertThat(existingIgnoresDefault.requiresRescale()).isTrue();
+ }
+
+ @Test
+ public void testDecideFixedBucketNumRejectsInvalidConfiguredDefault() {
+ Map optionMap = new HashMap<>();
+ optionMap.put(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM.key(), "0");
+
+ assertThatThrownBy(
+ () ->
+ PostponeUtils.decideFixedBucketNum(
+ 1, 0, null, CoreOptions.fromMap(optionMap)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM.key());
}
@Test
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
index e7dd850d634f..2f8f8959ed82 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
@@ -419,6 +419,8 @@ public void testPostponeMergeReadBuilderAndPushDown() throws Exception {
// A snapshot-bound builder must not pick up postpone files committed after its selection.
PostponeMergePlan realOnlyPlan = realOnlyBuilder.plan();
assertThat(realOnlyPlan.postponeSplits()).isEmpty();
+ assertThatThrownBy(() -> realOnlyPlan.bucketRouter().numBuckets(BinaryRow.singleColumn(1)))
+ .hasMessageContaining("Missing postpone bucket number");
assertThat(realOnlyPlan.realSplits())
.allSatisfy(split -> assertThat(split.snapshotId()).isEqualTo(realOnlySnapshotId));
@@ -436,7 +438,6 @@ public void testPostponeMergeReadBuilderAndPushDown() throws Exception {
.get()
.withFilter(valueFilter)
.withReadType(postponeTable.rowType().project("b"))
- .withDefaultBucketNum(1)
.plan();
assertThat(mergePlan.realSplits()).hasSize(3);
assertThat(mergePlan.postponeSplits()).hasSize(1);
@@ -523,8 +524,7 @@ public void testPostponeMergePlanAndRead() throws Exception {
PostponeMergeReadBuilder.create(postponeTable, null)
.get()
.withFilter(partitionFilter)
- .withReadType(postponeTable.rowType().project("b"))
- .withDefaultBucketNum(1);
+ .withReadType(postponeTable.rowType().project("b"));
PostponeMergePlan plan = readBuilder.plan();
assertThat(plan.realSplits()).hasSize(1);
@@ -574,26 +574,37 @@ public void testPostponeMergePlanPotentialBuckets() throws Exception {
postponeCommit.close();
PostponeMergeReadBuilder readBuilder =
- PostponeMergeReadBuilder.create(postponeTable, null).get().withDefaultBucketNum(1);
+ PostponeMergeReadBuilder.create(postponeTable, null).get();
PostponeMergePlan initialPlan = readBuilder.plan();
assertThat(initialPlan.numPotentialBuckets()).isEqualTo(3);
-
- PostponeMergePlan plan = readBuilder.reroute(initialPlan, 4);
-
- // Partition 1 uses its known bucket, partition 2 is real-only, and the new partition 3
- // may route to any of the four default buckets.
- assertThat(plan.numPotentialBuckets()).isEqualTo(6);
-
- FileStoreTable explicitDefaultTable =
+ assertThat(initialPlan.bucketRouter().numBuckets(BinaryRow.singleColumn(1))).isEqualTo(1);
+ assertThatThrownBy(() -> initialPlan.bucketRouter().numBuckets(BinaryRow.singleColumn(2)))
+ .hasMessageContaining("Missing postpone bucket number");
+
+ BinaryRow newPartition = BinaryRow.singleColumn(3);
+ long postponeFileSize =
+ initialPlan.postponeSplits().stream()
+ .filter(split -> split.partition().equals(newPartition))
+ .flatMap(split -> split.dataFiles().stream())
+ .mapToLong(DataFileMeta::fileSize)
+ .sum();
+ assertThat(postponeFileSize).isPositive();
+ FileStoreTable sizeEstimatedTable =
postponeTable.copy(
Collections.singletonMap(
- CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM.key(), "2"));
+ CoreOptions.POSTPONE_TARGET_SIZE_PER_BUCKET.key(), "1 b"));
+ PostponeMergePlan sizeEstimatedPlan =
+ PostponeMergeReadBuilder.create(sizeEstimatedTable, null).get().plan();
+ assertThat(sizeEstimatedPlan.numPotentialBuckets()).isEqualTo(2L + postponeFileSize);
+
+ Map explicitDefaultOptions = new HashMap<>();
+ explicitDefaultOptions.put(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM.key(), "2");
+ explicitDefaultOptions.put(CoreOptions.POSTPONE_TARGET_ROW_NUM_PER_BUCKET.key(), "100");
+ FileStoreTable explicitDefaultTable = postponeTable.copy(explicitDefaultOptions);
PostponeMergeReadBuilder explicitDefaultBuilder =
PostponeMergeReadBuilder.create(explicitDefaultTable, null).get();
PostponeMergePlan explicitDefaultPlan = explicitDefaultBuilder.plan();
assertThat(explicitDefaultPlan.numPotentialBuckets()).isEqualTo(4);
- assertThat(explicitDefaultBuilder.reroute(explicitDefaultPlan, 8).numPotentialBuckets())
- .isEqualTo(4);
}
@Test
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
index 501b6aed16e4..479468c42fcc 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
@@ -50,6 +50,7 @@
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.PostponeUtils;
import org.apache.paimon.table.PostponeUtils.CompactBucket;
+import org.apache.paimon.table.PostponeUtils.PostponeBucketNumResolver;
import org.apache.paimon.table.sink.ChannelComputer;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.InternalRowPartitionComputer;
@@ -317,20 +318,13 @@ protected boolean buildForPostponeBucketCompaction(
"Postpone bucket compaction currently does not support predicates");
Options options = new Options(table.options());
- int defaultBucketNum = options.get(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM);
- Optional targetRowNumPerBucket =
- options.getOptional(CoreOptions.POSTPONE_TARGET_ROW_NUM_PER_BUCKET);
Optional optionalSnapshot = table.latestSnapshot();
if (!optionalSnapshot.isPresent()) {
return buildNothingToCompact(env);
}
long snapshotId = optionalSnapshot.get().id();
- Map knownNumBuckets =
- PostponeUtils.getKnownNumBuckets(table, snapshotId);
- Map postponeRowCounts =
- targetRowNumPerBucket.isPresent()
- ? PostponeUtils.getPostponeRowCounts(table, snapshotId)
- : Collections.emptyMap();
+ PostponeBucketNumResolver bucketNumResolver =
+ PostponeUtils.createPostponeBucketNumResolver(table, snapshotId);
List postponePartitions =
table.newSnapshotReader()
@@ -359,15 +353,9 @@ protected boolean buildForPostponeBucketCompaction(
String commitUser = CoreOptions.createCommitUser(options);
List> dataStreams = new ArrayList<>();
for (BinaryRow partition : affectedPartitions) {
- int bucketNum =
- PostponeUtils.determineBucketNum(
- partition,
- knownNumBuckets,
- targetRowNumPerBucket,
- postponeRowCounts,
- defaultBucketNum);
+ int bucketNum = bucketNumResolver.numBuckets(partition);
FileStoreTable realTable =
- PostponeUtils.tableForPostponeCompact(table, bucketNum, snapshotId);
+ PostponeUtils.tableForPostponeRewrite(table, bucketNum, snapshotId);
LinkedHashMap partitionSpec =
partitionComputer.generatePartValues(partition);
@@ -419,8 +407,9 @@ protected boolean buildForPostponeBucketCompaction(
dataStreams.add(sourcePair.getRight());
}
+ int commitBucketNum = bucketNumResolver.numBuckets(affectedPartitions.iterator().next());
FileStoreTable fileStoreTable =
- PostponeUtils.tableForPostponeCompact(table, defaultBucketNum, snapshotId);
+ PostponeUtils.tableForPostponeRewrite(table, commitBucketNum, snapshotId);
FixedBucketSink sink = new FixedBucketSink(fileStoreTable, null);
DataStream dataStream = dataStreams.get(0);
for (int i = 1; i < dataStreams.size(); i++) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
index dca0f3b11935..eb3da3e16358 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
@@ -246,8 +246,7 @@ private DataStream buildStaticFileSource() {
}
int baseParallelism = basePostponeMergeParallelism();
- PostponeMergeReadBuilder readBuilder =
- optionalBuilder.get().withDefaultBucketNum(baseParallelism);
+ PostponeMergeReadBuilder readBuilder = optionalBuilder.get();
org.apache.paimon.types.RowType readType = projectedRowType();
if (readType != null) {
readBuilder.withReadType(readType);
@@ -257,9 +256,6 @@ private DataStream buildStaticFileSource() {
}
PostponeMergePlan plan = readBuilder.plan();
int mergeParallelism = inferPostponeMergeParallelism(plan, baseParallelism);
- if (mergeParallelism != baseParallelism) {
- plan = readBuilder.reroute(plan, mergeParallelism);
- }
return PostponeMergeOnRead.build(
env,
sourceName,
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java
index c3a020cbd372..6420bda34879 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java
@@ -560,6 +560,22 @@ public void testCompactWithTargetRowNumPerBucket() throws Exception {
"SELECT `partition`, COUNT(DISTINCT bucket) FROM `T$files` "
+ "GROUP BY `partition`")))
.containsExactlyInAnyOrder("+I[{0}, 1]", "+I[{1}, 3]");
+
+ tEnv.executeSql("ALTER TABLE T SET ('postpone.default-bucket-num' = '2')").await();
+ values.clear();
+ for (int j = 0; j < 450; j++) {
+ values.add(String.format("(2, %d, %d)", j, j));
+ }
+ tEnv.executeSql("INSERT INTO T VALUES " + String.join(", ", values)).await();
+ tEnv.executeSql("CALL sys.compact(`table` => 'default.T')").await();
+
+ // An explicitly configured default takes precedence over the row-count estimate of 3.
+ assertThat(
+ collect(
+ tEnv.executeSql(
+ "SELECT COUNT(DISTINCT bucket) FROM `T$files` "
+ + "WHERE `partition` = '{2}'")))
+ .containsExactly("+I[2]");
}
@Timeout(TIMEOUT)
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
index e3f70b81727b..dc71a9cfbfbf 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
@@ -159,7 +159,7 @@ abstract class PaimonBaseScan(table: InnerTable)
override def estimateStatistics: Statistics = {
if (postponeMergeOnRead.enabled) {
val splits =
- planPostponeMerge(SparkSession.active.sparkContext.defaultParallelism)
+ planPostponeMerge()
.map(_.corePlan.splits().asScala.toArray)
.getOrElse(Array.empty[Split])
PaimonStatistics(splits, readTableRowType, table.rowType(), table.statistics())
@@ -168,8 +168,8 @@ abstract class PaimonBaseScan(table: InnerTable)
}
}
- final private[spark] def planPostponeMerge(defaultBucketNum: Int): Option[MergePlan] = {
- postponeMergeOnRead.plan(defaultBucketNum)
+ final private[spark] def planPostponeMerge(): Option[MergePlan] = {
+ postponeMergeOnRead.plan()
}
override def supportedCustomMetrics: Array[CustomMetric] = {
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeOnRead.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeOnRead.scala
index 5d7ad0eb4ade..e340be28826a 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeOnRead.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeOnRead.scala
@@ -48,7 +48,7 @@ final private[spark] class PostponeMergeOnRead(scan: PaimonBaseScan) {
def enabled: Boolean = PostponeMergeOnRead.usesCustomSource(scan.table)
- def plan(defaultBucketNum: Int): Option[MergePlan] = synchronized {
+ def plan(): Option[MergePlan] = synchronized {
if (!enabled) {
return None
}
@@ -64,7 +64,6 @@ final private[spark] class PostponeMergeOnRead(scan: PaimonBaseScan) {
builder
.withReadType(scan.readTableRowType)
- .withDefaultBucketNum(defaultBucketNum)
.withMetricRegistry(scan.paimonMetricsRegistry)
if (scan.pushedDataFilters.nonEmpty) {
builder.withFilter(PredicateBuilder.and(scan.pushedDataFilters.toList.asJava))
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala
index 883a797ebe3a..a4c5babc0d94 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala
@@ -518,7 +518,7 @@ private[spark] class SparkPostponeStagedCommitter(
// per-partition totalBuckets values, so the copied table's uniform bucket number is only a
// compatibility fallback for the existing commit API.
val commitTable =
- PostponeUtils.tableForPostponeCompact(table, rescaleBucketNums.values.max, snapshotId)
+ PostponeUtils.tableForPostponeRewrite(table, rescaleBucketNums.values.max, snapshotId)
val commit = commitTable
.newCommit(fixedWriteCommitUser)
.appendCommitCheckConflict(true)
@@ -579,7 +579,7 @@ private[spark] class SparkPostponeStagedCommitter(
bucketNums.foreach {
case (partition, buckets) => javaBucketNums.put(partition, Integer.valueOf(buckets))
}
- PostponeUtils.createPostponeBucketRouter(table, javaBucketNums, 1)
+ PostponeUtils.createPostponeBucketRouter(table, javaBucketNums)
}
private def shuffleParallelism(bucketNums: Map[BinaryRow, Int]): Int = {
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
index ef49aab88333..b201b88110d9 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -19,9 +19,10 @@
package org.apache.paimon.spark.commands
import org.apache.paimon.{CoreOptions, Snapshot}
-import org.apache.paimon.CoreOptions.{PartitionSinkStrategy, WRITE_ONLY}
+import org.apache.paimon.CoreOptions.{COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT, PartitionSinkStrategy, WRITE_ONLY}
import org.apache.paimon.codegen.CodeGenUtils
import org.apache.paimon.crosspartition.{IndexBootstrap, KeyPartOrRow}
+import org.apache.paimon.data.BinaryRow
import org.apache.paimon.data.serializer.InternalSerializers
import org.apache.paimon.deletionvectors.DeletionVector
import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer
@@ -69,6 +70,7 @@ case class PaimonSparkWriter(
@transient private lazy val serializer = new CommitMessageSerializer
@transient private var stagedSparkSession: SparkSession = _
+ @transient private var directPostponeWriteBuilder: PostponeFixedBucketWriteBuilder = _
private var overwritePartitionSpec: Option[Map[String, String]] = None
private val writeType = {
@@ -92,6 +94,11 @@ case class PaimonSparkWriter(
Option(table.snapshotManager().latestSnapshot()).map(_.id())
else None
+ private val configuredPostponeDefaultBucketNum: Option[Int] = {
+ val bucketNum = coreOptions.postponeDefaultBucketNum()
+ if (bucketNum.isPresent) Some(bucketNum.get().intValue()) else None
+ }
+
val writeBuilder: BatchWriteBuilder = table.newBatchWriteBuilder()
def withOverwrite(): PaimonSparkWriter = withOverwrite(java.util.Collections.emptyMap())
@@ -121,6 +128,31 @@ case class PaimonSparkWriter(
val uriReaderFactory = uriReaderFactoryForBlobDescriptor
import sparkSession.implicits._
+ val directPostponeBucketNum =
+ if (
+ postponeBatchWriteFixedBucket && configuredPostponeDefaultBucketNum.isDefined &&
+ (overwritePartitionSpec.isDefined || baseSnapshotHasNoRealBuckets)
+ ) {
+ configuredPostponeDefaultBucketNum
+ } else {
+ None
+ }
+ val activeWriteBuilder: BatchWriteBuilder = directPostponeBucketNum match {
+ case Some(_) =>
+ val directWriteOptions = new java.util.HashMap[String, String]()
+ directWriteOptions.put(
+ COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(),
+ postponeBaseSnapshotId.getOrElse(0L).toString)
+ val builder = table.copy(directWriteOptions).newPostponeFixedBucketWriteBuilder()
+ overwritePartitionSpec.foreach(spec => builder.withOverwrite(spec.asJava))
+ directPostponeWriteBuilder = builder
+ builder
+ case None =>
+ directPostponeWriteBuilder = null
+ writeBuilder
+ }
+ stagedSparkSession = null
+
val withInitBucketCol = bucketMode match {
case BUCKET_UNAWARE => data
case KEY_DYNAMIC if !data.schema.fieldNames.contains(ROW_KIND_COL) =>
@@ -132,16 +164,18 @@ case class PaimonSparkWriter(
val rowKindColIdx = SparkRowUtils.getFieldIndex(withInitBucketCol.schema, ROW_KIND_COL)
val bucketColIdx = SparkRowUtils.getFieldIndex(withInitBucketCol.schema, BUCKET_COL)
val encoderGroupWithBucketCol = EncoderSerDeGroup(withInitBucketCol.schema)
+ val postponePartitionBucketComputer: Option[BinaryRow => Integer] =
+ directPostponeBucketNum.map(bucketNum => (_: BinaryRow) => Integer.valueOf(bucketNum))
def newWrite() =
PaimonDataWrite(
- writeBuilder,
+ activeWriteBuilder,
writeType,
rowKindColIdx,
writeRowTracking,
fullCompactionDeltaCommits,
batchId,
uriReaderFactory,
- None
+ postponePartitionBucketComputer
)
def sparkParallelism = {
@@ -291,6 +325,17 @@ case class PaimonSparkWriter(
)
}
+ case POSTPONE_MODE if directPostponeBucketNum.isDefined =>
+ // The configured bucket number is final for overwrite and for a table without real
+ // buckets, so route the input directly without first materializing bucket -2 files.
+ writeWithBucketProcessor(
+ withInitBucketCol,
+ PostponeFixBucketProcessor(
+ table,
+ bucketColIdx,
+ encoderGroupWithBucketCol,
+ postponePartitionBucketComputer.get))
+
case BUCKET_UNAWARE | POSTPONE_MODE =>
var input = data
if (tableSchema.partitionKeys().size() > 0) {
@@ -341,7 +386,7 @@ case class PaimonSparkWriter(
}
val taskResults = written.collect().toSeq
- if (postponeBatchWriteFixedBucket) {
+ if (postponeBatchWriteFixedBucket && directPostponeWriteBuilder == null) {
stagedSparkSession = sparkSession
}
WriteTaskResult.merge(taskResults)
@@ -419,7 +464,7 @@ case class PaimonSparkWriter(
}
def commit(commitMessages: Seq[CommitMessage], operation: Snapshot.Operation): Unit = {
- if (postponeBatchWriteFixedBucket) {
+ if (postponeBatchWriteFixedBucket && directPostponeWriteBuilder == null) {
if (stagedSparkSession == null) {
throw new IllegalStateException("Postpone staged write has no SparkSession.")
}
@@ -432,7 +477,9 @@ case class PaimonSparkWriter(
postCommit(finalMessages)
return
}
- val tableCommit = writeBuilder.newCommit()
+ val activeWriteBuilder =
+ Option(directPostponeWriteBuilder).getOrElse(writeBuilder)
+ val tableCommit = activeWriteBuilder.newCommit()
if (operation != null) {
tableCommit.withOperation(operation)
}
@@ -446,6 +493,18 @@ case class PaimonSparkWriter(
postCommit(commitMessages)
}
+ private def baseSnapshotHasNoRealBuckets: Boolean = {
+ postponeBaseSnapshotId.forall {
+ snapshotId =>
+ !table
+ .newSnapshotReader()
+ .withSnapshot(snapshotId)
+ .onlyReadRealBuckets()
+ .readFileIterator()
+ .hasNext
+ }
+ }
+
/** Bootstrap and repartition for cross partition mode. */
private def bootstrapAndRepartitionByKeyHash(
data: DataFrame,
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
index 53a10b02870d..770b9eec66cd 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
@@ -86,7 +86,7 @@ case class PaimonStrategy(spark: SparkSession)
case PhysicalOperation(projects, filters, relation: DataSourceV2ScanRelation) =>
relation.scan match {
case scan: PaimonScan if PostponeMergeOnRead.usesCustomSource(scan.table) =>
- scan.planPostponeMerge(spark.sparkContext.defaultParallelism) match {
+ scan.planPostponeMerge() match {
case Some(mergePlan) =>
val inputScan = PostponeMergeInputScan(mergePlan)
val inputOutput =
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala
index 25d4f8a91d18..a39636e9c15f 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala
@@ -29,7 +29,7 @@ import org.apache.paimon.spark.schema.SparkSystemColumns.{BUCKET_COL, ROW_KIND_C
import org.apache.paimon.spark.util.{ScanPlanHelper, SparkRowUtils}
import org.apache.paimon.spark.write.{PaimonDataWrite, WriteTaskResult}
import org.apache.paimon.table.{BlobDescriptorReaderFactory, BucketMode, FileStoreTable, PostponeUtils}
-import org.apache.paimon.table.PostponeUtils.PostponeBucketAssigner
+import org.apache.paimon.table.PostponeUtils.PostponeBucketNumResolver
import org.apache.paimon.table.sink.{CommitMessage, CommitMessageImpl}
import org.apache.paimon.utils.{SerializationUtils, UriReaderFactory}
@@ -58,17 +58,10 @@ case class SparkPostponeCompactProcedure(
@transient relation: DataSourceV2Relation) {
private val LOG = LoggerFactory.getLogger(getClass)
- private def createPostponeBucketAssigner(snapshotId: Long) = {
- PostponeUtils.createPostponeBucketAssigner(
- table,
- snapshotId,
- spark.sparkContext.defaultParallelism)
- }
-
private def newDataWrite(
realTable: FileStoreTable,
rowKindColIdx: Int,
- bucketAssigner: PostponeBucketAssigner,
+ bucketNumResolver: PostponeBucketNumResolver,
uriReaderFactoryForBlobDescriptor: UriReaderFactory): PaimonDataWrite = {
val rowType = table.rowType()
val coreOptions = table.coreOptions()
@@ -81,7 +74,7 @@ case class SparkPostponeCompactProcedure(
Option.apply(coreOptions.fullCompactionDeltaCommits()),
None,
uriReaderFactoryForBlobDescriptor,
- Some(partition => bucketAssigner.assign(partition))
+ Some(partition => bucketNumResolver.numBuckets(partition))
)
dataWrite
}
@@ -107,8 +100,8 @@ case class SparkPostponeCompactProcedure(
return
}
val snapshotId = snapshot.id()
- val bucketAssigner = createPostponeBucketAssigner(snapshotId)
- val realTable = PostponeUtils.tableForPostponeCompact(table, 1, snapshotId)
+ val bucketNumResolver = PostponeUtils.createPostponeBucketNumResolver(table, snapshotId)
+ val realTable = PostponeUtils.tableForPostponeRewrite(table, 1, snapshotId)
// Read data splits from the POSTPONE_BUCKET (-2)
val splits =
@@ -147,7 +140,7 @@ case class SparkPostponeCompactProcedure(
table,
bucketColIdx,
encoderGroupWithBucketCol,
- partition => bucketAssigner.assign(partition)
+ partition => bucketNumResolver.numBuckets(partition)
)
val dataFrame = withInitBucketCol
.mapPartitions(processor.processPartition)(encoderGroupWithBucketCol.encoder)
@@ -205,7 +198,7 @@ case class SparkPostponeCompactProcedure(
Iterator.empty
} else {
val dataWrite =
- newDataWrite(realTable, rowWorkAndKind._2, bucketAssigner, uriReaderFactory)
+ newDataWrite(realTable, rowWorkAndKind._2, bucketNumResolver, uriReaderFactory)
dataWrite.write.withWriteRestore(
new FileSystemWriteRestore(
realTable.coreOptions(),
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala
index 2f59dd9e0cfd..5ad96cba0e2e 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala
@@ -102,6 +102,139 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
}
}
+ test("Postpone bucket table: configured default directly writes overwrite") {
+ Seq(
+ (
+ "static",
+ """
+ |INSERT OVERWRITE t SELECT
+ |CAST(id AS INT) AS k,
+ |CAST(id AS STRING) AS v,
+ |0 AS pt
+ |FROM range(100, 200)
+ |""".stripMargin,
+ false),
+ (
+ "static",
+ """
+ |INSERT OVERWRITE t PARTITION (pt = 0) SELECT
+ |CAST(id AS INT) AS k,
+ |CAST(id AS STRING) AS v
+ |FROM range(100, 200)
+ |""".stripMargin,
+ true),
+ (
+ "dynamic",
+ """
+ |INSERT OVERWRITE t SELECT
+ |CAST(id AS INT) AS k,
+ |CAST(id AS STRING) AS v,
+ |0 AS pt
+ |FROM range(100, 200)
+ |""".stripMargin,
+ true)
+ ).foreach {
+ case (partitionOverwriteMode, overwriteSql, preservesPartitionOne) =>
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING,
+ | pt INT
+ |) PARTITIONED BY (pt)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'k, pt',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'true',
+ | 'postpone.target-row-num-per-bucket' = '1'
+ |)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO t
+ |SELECT CAST(id AS INT), CAST(id AS STRING), 0 AS pt FROM range(0, 16)
+ |UNION ALL SELECT 1000, 'untouched-real', 1
+ |""".stripMargin)
+ val initialBuckets = PostponeUtils.getKnownNumBuckets(loadTable("t"))
+ assert(initialBuckets.get(BinaryRow.singleColumn(0)) == 16)
+ assert(initialBuckets.get(BinaryRow.singleColumn(1)) == 1)
+
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" -> "false") {
+ sql("INSERT INTO t VALUES (2000, 'overwritten-postpone', 0), (2001, 'untouched-postpone', 1)")
+ }
+ sql("ALTER TABLE t SET TBLPROPERTIES ('postpone.default-bucket-num' = '3')")
+
+ withSparkSQLConf(
+ "spark.sql.adaptive.enabled" -> "false",
+ "spark.sql.sources.partitionOverwriteMode" -> partitionOverwriteMode) {
+ val jobs = countSparkJobs("postpone-default-overwrite") {
+ sql(overwriteSql)
+ }
+ assert(jobs == 1, s"Direct overwrite should use one Spark job, but found $jobs.")
+ }
+
+ val resultBuckets = PostponeUtils.getKnownNumBuckets(loadTable("t"))
+ assert(resultBuckets.get(BinaryRow.singleColumn(0)) == 3)
+ assert(resultBuckets.containsKey(BinaryRow.singleColumn(1)) == preservesPartitionOne)
+ val retainedPartitionCount = if (preservesPartitionOne) 1L else 0L
+ checkAnswer(sql("SELECT count(*) FROM t WHERE pt = 1"), Seq(Row(retainedPartitionCount)))
+ checkAnswer(
+ sql("SELECT count(*) FROM `t$buckets` WHERE bucket = -2"),
+ Seq(Row(retainedPartitionCount)))
+ }
+ }
+ }
+
+ test("Postpone bucket table: configured default directly writes without real buckets") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING,
+ | pt INT
+ |) PARTITIONED BY (pt)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'k, pt',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'true',
+ | 'postpone.default-bucket-num' = '3',
+ | 'postpone.target-row-num-per-bucket' = '1'
+ |)
+ |""".stripMargin)
+
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" -> "false") {
+ sql("INSERT INTO t VALUES (0, 'historical-postpone', 0)")
+ }
+ assert(PostponeUtils.getKnownNumBuckets(loadTable("t")).isEmpty)
+
+ withSparkSQLConf("spark.sql.adaptive.enabled" -> "false") {
+ val jobs = countSparkJobs("postpone-default-new-layout") {
+ sql("INSERT INTO t VALUES (1, 'p0-real', 0), (2, 'p1-real', 1)")
+ }
+ assert(jobs == 1, s"Direct new-layout write should use one Spark job, but found $jobs.")
+ }
+
+ val initialBuckets = PostponeUtils.getKnownNumBuckets(loadTable("t"))
+ assert(initialBuckets.get(BinaryRow.singleColumn(0)) == 3)
+ assert(initialBuckets.get(BinaryRow.singleColumn(1)) == 3)
+ checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket = -2"), Seq(Row(1L)))
+
+ sql("ALTER TABLE t SET TBLPROPERTIES ('postpone.default-bucket-num' = '5')")
+ withSparkSQLConf("spark.sql.adaptive.enabled" -> "false") {
+ val jobs = countSparkJobs("postpone-default-mixed-layout") {
+ sql("INSERT INTO t VALUES (3, 'p0-append', 0), (4, 'p2-new', 2)")
+ }
+ assert(jobs == 2, s"Mixed existing/new write should stay staged, but found $jobs jobs.")
+ }
+
+ val resultBuckets = PostponeUtils.getKnownNumBuckets(loadTable("t"))
+ assert(resultBuckets.get(BinaryRow.singleColumn(0)) == 3)
+ assert(resultBuckets.get(BinaryRow.singleColumn(1)) == 3)
+ assert(resultBuckets.get(BinaryRow.singleColumn(2)) == 5)
+ checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket = -2"), Seq(Row(1L)))
+ }
+ }
+
test("Postpone bucket table: staged rescale supports per-partition layouts") {
withTable("t") {
sql("""
@@ -899,7 +1032,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
case relation: DataSourceV2ScanRelation if relation.scan.isInstanceOf[PaimonScan] =>
relation.scan.asInstanceOf[PaimonScan]
}.get
- assert(postponeScan.planPostponeMerge(spark.sparkContext.defaultParallelism).isDefined)
+ assert(postponeScan.planPostponeMerge().isDefined)
assert(postponeScan.filterAttributes().isEmpty)
assert(
intercept[UnsupportedOperationException](postponeScan.toBatch).getMessage
@@ -925,7 +1058,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
relation.scan.asInstanceOf[PaimonScan]
}.get
val realSplits = mergeScan
- .planPostponeMerge(spark.sparkContext.defaultParallelism)
+ .planPostponeMerge()
.get
.corePlan
.realSplits()
@@ -1294,7 +1427,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
}.get
assert(
pinnedRealScan
- .planPostponeMerge(spark.sparkContext.defaultParallelism)
+ .planPostponeMerge()
.isDefined)
withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" -> "false") {
@@ -1310,7 +1443,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
}.get
assert(
pinnedEmptyScan
- .planPostponeMerge(spark.sparkContext.defaultParallelism)
+ .planPostponeMerge()
.isEmpty)
sql("INSERT INTO empty_t VALUES (1, 'committed-later')")
@@ -1468,14 +1601,15 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
)
sql("SET spark.default.parallelism = 2")
- // compact
+ // Compact estimates one logical bucket from the default target size; Spark parallelism is
+ // only an execution setting.
sql("CALL sys.compact(table => 't')")
checkAnswer(sql("SELECT count(*) FROM t"), Seq(Row(1000)))
checkAnswer(sql("SELECT sum(k) FROM t"), Seq(Row((0 until 1000).sum)))
checkAnswer(
sql("SELECT distinct(bucket) FROM `t$buckets` ORDER BY bucket"),
- Seq(Row(0), Row(1))
+ Seq(Row(0))
)
}
}
@@ -1687,6 +1821,16 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
.sum
}
+ private def countSparkJobs(groupPrefix: String)(action: => Unit): Int = {
+ val jobGroup = s"$groupPrefix-${System.nanoTime()}"
+ spark.sparkContext.setJobGroup(jobGroup, jobGroup)
+ try {
+ action
+ } finally {
+ spark.sparkContext.clearJobGroup()
+ }
+ spark.sparkContext.statusTracker.getJobIdsForGroup(jobGroup).length
+ }
}
object PostponeBucketTableTest {