This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 50bfb69cc6 [core][flink][spark] Support default bucket number for
postpone writes (#9067)
50bfb69cc6 is described below
commit 50bfb69cc695e757ed4c5d949512610d755361a0
Author: Zouxxyy <[email protected]>
AuthorDate: Thu Aug 6 22:14:58 2026 +0800
[core][flink][spark] Support default bucket number for postpone writes
(#9067)
---
docs/docs/primary-key-table/data-distribution.md | 37 +--
docs/generated/core_configuration.html | 8 +-
.../main/java/org/apache/paimon/CoreOptions.java | 19 +-
.../org/apache/paimon/table/PostponeUtils.java | 274 +++++++++++----------
.../paimon/table/source/PostponeMergePlan.java | 10 -
.../table/source/PostponeMergeReadBuilder.java | 45 ++--
.../org/apache/paimon/table/PostponeUtilsTest.java | 83 +++++--
.../apache/paimon/table/source/TableScanTest.java | 41 +--
.../apache/paimon/flink/action/CompactAction.java | 25 +-
.../paimon/flink/source/FlinkSourceBuilder.java | 6 +-
.../paimon/flink/PostponeBucketTableITCase.java | 16 ++
.../org/apache/paimon/spark/PaimonBaseScan.scala | 6 +-
.../apache/paimon/spark/PostponeMergeOnRead.scala | 3 +-
.../spark/SparkPostponeStagedCommitter.scala | 4 +-
.../paimon/spark/commands/PaimonSparkWriter.scala | 71 +++++-
.../paimon/spark/execution/PaimonStrategy.scala | 2 +-
.../procedure/SparkPostponeCompactProcedure.scala | 21 +-
.../paimon/spark/sql/PostponeBucketTableTest.scala | 156 +++++++++++-
18 files changed, 552 insertions(+), 275 deletions(-)
diff --git a/docs/docs/primary-key-table/data-distribution.md
b/docs/docs/primary-key-table/data-distribution.md
index b7a0fecb7c..360febf904 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 0c0c1a0a8d..a7c8114dd8 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1316,9 +1316,9 @@ For an internal format table in a REST catalog, it also
makes the catalog own th
</tr>
<tr>
<td><h5>postpone.default-bucket-num</h5></td>
- <td style="word-wrap: break-word;">1</td>
+ <td style="word-wrap: break-word;">(none)</td>
<td>Integer</td>
- <td>Bucket number for the partitions compacted for the first time
in postpone bucket tables.</td>
+ <td>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.</td>
</tr>
<tr>
<td><h5>postpone.merge-on-read</h5></td>
@@ -1330,13 +1330,13 @@ For an internal format table in a REST catalog, it also
makes the catalog own th
<td><h5>postpone.target-row-num-per-bucket</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>Long</td>
- <td>Target row number per bucket when estimating the required
bucket number from the current staged batch or compacting postpone bucket
files.</td>
+ <td>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'.</td>
</tr>
<tr>
<td><h5>postpone.target-size-per-bucket</h5></td>
<td style="word-wrap: break-word;">1 gb</td>
<td>MemorySize</td>
- <td>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.</td>
+ <td>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.</td>
</tr>
<tr>
<td><h5>primary-key</h5></td>
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 96fdbf731b..3be4d072e1 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 class CoreOptions implements Serializable {
public static final ConfigOption<Integer> 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<Long> 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<MemorySize>
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<Long> GLOBAL_INDEX_ROW_COUNT_PER_SHARD =
@@ -4465,8 +4465,15 @@ public class CoreOptions implements Serializable {
return
options.get(POSTPONE_BATCH_WRITE_FIXED_BUCKET_RESCALE_LOAD_FACTOR);
}
- public int postponeDefaultBucketNum() {
- return options.get(POSTPONE_DEFAULT_BUCKET_NUM);
+ public Optional<Integer> postponeDefaultBucketNum() {
+ Optional<Integer> 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<Long> 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 0d2a77ad0b..7ea59f78a9 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 class PostponeUtils {
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<BinaryRow, Integer> knownNumBuckets =
- getKnownNumBuckets(table, snapshotId, partitionFilter);
- Map<BinaryRow, Long> 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<BinaryRow>
postponePartitions) {
+ CoreOptions options = table.coreOptions();
+ Map<BinaryRow, Integer> numBucketsByPartition =
+ postponePartitions == null
+ ? getKnownNumBuckets(table, snapshotId)
+ : getKnownNumBuckets(table, snapshotId,
postponePartitions);
+ Integer configuredDefaultBucketNum =
options.postponeDefaultBucketNum().orElse(null);
+ if (configuredDefaultBucketNum == null) {
+ Optional<Long> 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<BinaryRow>
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<BinaryRow, Integer> numBucketsByPartition,
- int defaultBucketNum) {
- checkArgument(defaultBucketNum > 0, "Default postpone bucket number
must be positive.");
+ FileStoreTable table, Map<BinaryRow, Integer>
numBucketsByPartition) {
Map<BinaryRow, Integer> copied = new HashMap<>();
for (Map.Entry<BinaryRow, Integer> entry :
numBucketsByPartition.entrySet()) {
checkArgument(
@@ -179,13 +187,11 @@ public class PostponeUtils {
"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<String> trimmedPrimaryKeys = table.schema().trimmedPrimaryKeys();
int[] bucketKeyMapping =
table.schema().bucketKeys().stream()
@@ -202,7 +208,7 @@ public class PostponeUtils {
PrimaryKeyTableUtils.PrimaryKeyFieldsExtractor.EXTRACTOR.keyFields(
table.schema()));
return new PostponeBucketRouter(
- bucketAssigner,
+ bucketNumResolver,
keyType,
table.schema().logicalBucketKeyType(),
bucketKeyMapping,
@@ -210,52 +216,57 @@ public class PostponeUtils {
}
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<BinaryRow, Integer> knownNumBuckets,
- Optional<Long> targetRowNumPerBucket,
- Map<BinaryRow, Long> postponeRowCounts,
- int defaultBucketNum) {
- return determineBucketNum(
- partition,
- knownNumBuckets,
- targetRowNumPerBucket.orElse(null),
- postponeRowCounts,
- defaultBucketNum);
- }
-
- public static int determineBucketNum(
+ private static void addEstimatedBucketNums(
+ Map<BinaryRow, Integer> numBucketsByPartition,
+ Map<BinaryRow, Long> valuesByPartition,
+ long targetValuePerBucket,
+ String targetOptionKey) {
+ for (Map.Entry<BinaryRow, Long> entry : valuesByPartition.entrySet()) {
+ if (!numBucketsByPartition.containsKey(entry.getKey())) {
+ numBucketsByPartition.put(
+ entry.getKey(),
+ computeBucketNum(entry.getValue(),
targetValuePerBucket, targetOptionKey));
+ }
+ }
+ }
+
+ static int determineBucketNum(
BinaryRow partition,
- Map<BinaryRow, Integer> knownNumBuckets,
- @Nullable Long targetRowNumPerBucket,
- Map<BinaryRow, Long> 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<BinaryRow, Integer> 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 class PostponeUtils {
existingBucketNum == null || existingBucketNum > 0,
"Existing bucket number must be positive.");
+ Optional<Integer> 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 @@ public class PostponeUtils {
return rowCounts;
}
- public static FileStoreTable tableForPostponeCompact(
+ private static Iterator<ManifestEntry> postponeFileIterator(
+ FileStoreTable table, long snapshotId, @Nullable List<BinaryRow>
postponePartitions) {
+ SnapshotReader reader =
+ table.newSnapshotReader()
+ .withSnapshot(snapshotId)
+ .withBucket(BucketMode.POSTPONE_BUCKET);
+ if (postponePartitions != null) {
+ reader.withPartitionFilter(postponePartitions);
+ }
+ return reader.readFileIterator();
+ }
+
+ static Map<BinaryRow, Long> 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<BinaryRow, Long>
getPostponeFileSizes(Iterator<ManifestEntry> iterator) {
+ Map<BinaryRow, Long> 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<String, String> 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<String, String> 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<BinaryRow, Integer> knownNumBuckets;
- @Nullable private final Long targetRowNumPerBucket;
- private final Map<BinaryRow, Long> postponeRowCounts;
- private final int defaultBucketNum;
+ private final Map<BinaryRow, Integer> numBucketsByPartition;
+ @Nullable private final Integer configuredDefaultBucketNum;
- private PostponeBucketAssigner(
- Map<BinaryRow, Integer> knownNumBuckets,
- @Nullable Long targetRowNumPerBucket,
- Map<BinaryRow, Long> postponeRowCounts,
- int defaultBucketNum) {
- this.knownNumBuckets = knownNumBuckets;
- this.targetRowNumPerBucket = targetRowNumPerBucket;
- this.postponeRowCounts = postponeRowCounts;
- this.defaultBucketNum = defaultBucketNum;
+ private PostponeBucketNumResolver(
+ Map<BinaryRow, Integer> 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 class PostponeUtils {
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 class PostponeUtils {
@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 class PostponeUtils {
}
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 a65e415ee0..b2e6bc460e 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 final class PostponeMergePlan implements
TableScan.Plan {
return numPotentialBuckets;
}
- PostponeMergePlan withDefaultBucketNum(int newDefaultBucketNum) {
- return new PostponeMergePlan(
- realSplits,
- postponeSplits,
- bucketRouter.withDefaultBucketNum(newDefaultBucketNum),
- keyType,
- resultReadType,
- mergeReadType);
- }
-
private static long numPotentialBuckets(
List<DataSplit> realSplits,
List<DataSplit> 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 c0561c09db..147d12e05a 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 @@ package org.apache.paimon.table.source;
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 javax.annotation.Nullable;
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 final class PostponeMergeReadBuilder implements
Serializable {
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 final class PostponeMergeReadBuilder implements
Serializable {
postponeReader.withPartitionFilter(partitionFilter);
}
+ List<DataSplit> realSplits = realReader.read().dataSplits();
+ List<DataSplit> postponeSplits =
+
PostponeUtils.groupPostponeFiles(postponeReader.read().dataSplits());
+ PostponeUtils.PostponeBucketRouter bucketRouter;
+ if (postponeSplits.isEmpty()) {
+ bucketRouter = PostponeUtils.createPostponeBucketRouter(table,
Collections.emptyMap());
+ } else {
+ List<BinaryRow> 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 final class PostponeMergeReadBuilder implements
Serializable {
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 1d07359b6d..d6c6ddc5d3 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 class PostponeUtilsTest {
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 class PostponeUtilsTest {
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<Map<String, String>> options =
ArgumentCaptor.forClass(Map.class);
@@ -262,33 +284,24 @@ public class PostponeUtilsTest {
@Test
public void testDetermineBucketNum() {
- Map<BinaryRow, Integer> knownNumBuckets = new HashMap<>();
- Map<BinaryRow, Long> postponeRowCounts = new HashMap<>();
+ Map<BinaryRow, Integer> 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 class PostponeUtilsTest {
29, 0, 7, CoreOptions.fromMap(lowerLoadFactorOptions));
assertThat(lowerLoadFactor.targetBucketNum()).isEqualTo(16);
assertThat(lowerLoadFactor.requiresRescale()).isTrue();
+
+ Map<String, String> 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<String, String> 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 e7dd850d63..2f8f8959ed 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 class TableScanTest extends ScannerTestBase {
// 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 class TableScanTest extends ScannerTestBase {
.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 class TableScanTest extends ScannerTestBase {
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 class TableScanTest extends ScannerTestBase {
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<String, String> 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 501b6aed16..479468c42f 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.BucketMode;
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 @@ public class CompactAction extends TableActionBase {
"Postpone bucket compaction currently does not support
predicates");
Options options = new Options(table.options());
- int defaultBucketNum =
options.get(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM);
- Optional<Long> targetRowNumPerBucket =
-
options.getOptional(CoreOptions.POSTPONE_TARGET_ROW_NUM_PER_BUCKET);
Optional<Snapshot> optionalSnapshot = table.latestSnapshot();
if (!optionalSnapshot.isPresent()) {
return buildNothingToCompact(env);
}
long snapshotId = optionalSnapshot.get().id();
- Map<BinaryRow, Integer> knownNumBuckets =
- PostponeUtils.getKnownNumBuckets(table, snapshotId);
- Map<BinaryRow, Long> postponeRowCounts =
- targetRowNumPerBucket.isPresent()
- ? PostponeUtils.getPostponeRowCounts(table, snapshotId)
- : Collections.emptyMap();
+ PostponeBucketNumResolver bucketNumResolver =
+ PostponeUtils.createPostponeBucketNumResolver(table,
snapshotId);
List<BinaryRow> postponePartitions =
table.newSnapshotReader()
@@ -359,15 +353,9 @@ public class CompactAction extends TableActionBase {
String commitUser = CoreOptions.createCommitUser(options);
List<DataStream<Committable>> 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<String, String> partitionSpec =
partitionComputer.generatePartValues(partition);
@@ -419,8 +407,9 @@ public class CompactAction extends TableActionBase {
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<Committable> 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 dca0f3b119..eb3da3e163 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 @@ public class FlinkSourceBuilder {
}
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 @@ public class FlinkSourceBuilder {
}
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 c3a020cbd3..6420bda348 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 class PostponeBucketTableITCase extends
AbstractTestBase {
"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 e3f70b8172..dc71a9cfbf 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 5d7ad0eb4a..e340be2882 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 883a797ebe..a4c5babc0d 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 ef49aab883..b201b88110 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 53a10b0287..770b9eec66 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 25d4f8a91d..a39636e9c1 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 2f59dd9e0c..5ad96cba0e 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 {
