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 a8cea0922b [spark] Stage postpone fixed-bucket writes before rescale
(#9017)
a8cea0922b is described below
commit a8cea0922bee3679f03c4d6e7de6da56c36d9b8b
Author: Zouxxyy <[email protected]>
AuthorDate: Wed Aug 5 15:39:11 2026 +0800
[spark] Stage postpone fixed-bucket writes before rescale (#9017)
---
docs/docs/primary-key-table/data-distribution.md | 40 +-
docs/generated/core_configuration.html | 12 +-
.../main/java/org/apache/paimon/CoreOptions.java | 19 +-
.../org/apache/paimon/table/PostponeUtils.java | 164 ++++-
.../table/source/PostponeMergeReadBuilder.java | 28 +-
.../org/apache/paimon/table/PostponeUtilsTest.java | 101 +++
.../paimon/flink/PostponeBucketTableITCase.java | 37 ++
.../procedure/CompactChainTableProcedure.java | 2 +-
.../paimon/spark/procedure/CompactProcedure.java | 2 +-
.../paimon/spark/procedure/RescaleProcedure.java | 2 +-
.../paimon/spark/PostponeMergeInputScan.scala | 2 +-
.../spark/SparkPostponeStagedCommitter.scala | 717 +++++++++++++++++++++
.../paimon/spark/commands/PaimonSparkWriter.scala | 265 ++------
.../spark/commands/WriteIntoPaimonTable.scala | 7 +-
.../procedure/SparkPostponeCompactProcedure.scala | 1 +
.../spark/procedure/RescaleProcedureTest.scala | 51 ++
.../paimon/spark/sql/PostponeBucketTableTest.scala | 642 ++++++++++++++++--
17 files changed, 1762 insertions(+), 330 deletions(-)
diff --git a/docs/docs/primary-key-table/data-distribution.md
b/docs/docs/primary-key-table/data-distribution.md
index a7de95c838..b7a0fecb7c 100644
--- a/docs/docs/primary-key-table/data-distribution.md
+++ b/docs/docs/primary-key-table/data-distribution.md
@@ -70,22 +70,30 @@ 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, batch writes set `postpone.batch-write-fixed-bucket` to `true`
-and write records directly to real buckets.
-For a Spark batch write to a partition without real bucket files,
-Spark calculates the bucket number from the uncompressed Paimon `BinaryRow`
serialized size.
-The target size is configured by `postpone.target-size-per-bucket` and
defaults to `1 GB`.
-Existing postpone files do not record their uncompressed serialized size, so
Spark estimates their
-size using their row count and the average serialized size of incoming rows in
the same partition.
-You can instead configure `postpone.target-row-num-per-bucket` to calculate
the bucket number
-from row counts; this option takes precedence over the target size.
-The calculated bucket number is at least `1` and is limited by
-`postpone.batch-write-fixed-bucket.max-parallelism`.
-The serialized size is measured before file encoding and compression, so it is
not the exact
-ORC or Parquet size on storage.
-Inferring the data amount adds a statistics stage; Spark caches the batch so
that inference and
-writing use the same input rows. Spark skips this stage for an unpartitioned
table that already
-has real bucket files, or when
`postpone.batch-write-fixed-bucket.max-parallelism` is `1`.
+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:
+
+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
+ `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.
+
+An existing partition normally keeps its bucket number. Spark first rescales
its real buckets when
+the uncapped required bucket number is greater than the existing bucket number
multiplied by
+`postpone.batch-write-fixed-bucket.rescale-load-factor` (default `32`), and
the capped result is
+larger than the existing layout. Different partitions may have different
target bucket numbers.
+The rescale is a separate overwrite commit which changes real buckets only;
the current batch is
+appended in the following commit.
+
+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.
When `postpone.batch-write-fixed-bucket` is `false`,
records are first stored in the `bucket-postpone` directory of each partition
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 78d3e5a168..bb5fba90d3 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1306,7 +1306,13 @@ For an internal format table in a REST catalog, it also
makes the catalog own th
<td><h5>postpone.batch-write-fixed-bucket.max-parallelism</h5></td>
<td style="word-wrap: break-word;">2048</td>
<td>Integer</td>
- <td>The number of partitions for global index.</td>
+ <td>Maximum bucket number inferred for a partition by a
fixed-bucket batch write. The inferred number is rounded up to a power of two
before applying this limit.</td>
+ </tr>
+ <tr>
+
<td><h5>postpone.batch-write-fixed-bucket.rescale-load-factor</h5></td>
+ <td style="word-wrap: break-word;">32</td>
+ <td>Integer</td>
+ <td>Maximum tolerated ratio between the required bucket number and
the existing bucket number before a fixed-bucket batch write enlarges the
existing layout. Rescaling also requires the configured maximum parallelism to
permit a larger bucket number.</td>
</tr>
<tr>
<td><h5>postpone.default-bucket-num</h5></td>
@@ -1324,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 batch writing fixed buckets
or compacting postpone bucket files for a partition without real bucket
data.</td>
+ <td>Target row number per bucket when estimating the required
bucket number from the current staged batch or compacting postpone bucket
files.</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 uncompressed serialized data size per bucket when Spark
batch writes fixed buckets for a partition without real bucket data. This
option is ignored when 'postpone.target-row-num-per-bucket' is configured.</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>
</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 a405d1318e..2454b84c80 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2752,7 +2752,16 @@ public class CoreOptions implements Serializable {
key("postpone.batch-write-fixed-bucket.max-parallelism")
.intType()
.defaultValue(2048)
- .withDescription("The number of partitions for global
index.");
+ .withDescription(
+ "Maximum bucket number inferred for a partition by
a fixed-bucket batch write. The inferred number is rounded up to a power of two
before applying this limit.");
+
+ public static final ConfigOption<Integer>
+ POSTPONE_BATCH_WRITE_FIXED_BUCKET_RESCALE_LOAD_FACTOR =
+
key("postpone.batch-write-fixed-bucket.rescale-load-factor")
+ .intType()
+ .defaultValue(32)
+ .withDescription(
+ "Maximum tolerated ratio between the
required bucket number and the existing bucket number before a fixed-bucket
batch write enlarges the existing layout. Rescaling also requires the
configured maximum parallelism to permit a larger bucket number.");
public static final ConfigOption<Integer> POSTPONE_DEFAULT_BUCKET_NUM =
key("postpone.default-bucket-num")
@@ -2766,14 +2775,14 @@ public class CoreOptions implements Serializable {
.longType()
.noDefaultValue()
.withDescription(
- "Target row number per bucket when batch writing
fixed buckets or compacting postpone bucket files for a partition without real
bucket data.");
+ "Target row number per bucket when estimating the
required bucket number from the current staged batch or compacting postpone
bucket files.");
public static final ConfigOption<MemorySize>
POSTPONE_TARGET_SIZE_PER_BUCKET =
key("postpone.target-size-per-bucket")
.memoryType()
.defaultValue(MemorySize.parse("1 gb"))
.withDescription(
- "Target uncompressed serialized data size per
bucket when Spark batch writes fixed buckets for a partition without real
bucket data. "
+ "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.");
public static final ConfigOption<Long> GLOBAL_INDEX_ROW_COUNT_PER_SHARD =
@@ -4452,6 +4461,10 @@ public class CoreOptions implements Serializable {
return options.get(POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM);
}
+ public int postponeBatchWriteFixedBucketRescaleLoadFactor() {
+ return
options.get(POSTPONE_BATCH_WRITE_FIXED_BUCKET_RESCALE_LOAD_FACTOR);
+ }
+
public int postponeDefaultBucketNum() {
return options.get(POSTPONE_DEFAULT_BUCKET_NUM);
}
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 60b464fd15..0d2a77ad0b 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
@@ -39,6 +39,7 @@ import org.apache.paimon.utils.Pair;
import javax.annotation.Nullable;
import java.io.Serializable;
+import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
@@ -128,10 +129,10 @@ public class PostponeUtils {
public static PostponeBucketAssigner createPostponeBucketAssigner(
FileStoreTable table, long snapshotId, int defaultParallelism) {
- return createPostponeBucketAssigner(table, snapshotId,
defaultParallelism, null);
+ return loadPostponeBucketAssigner(table, snapshotId,
defaultParallelism, null);
}
- private static PostponeBucketAssigner createPostponeBucketAssigner(
+ private static PostponeBucketAssigner loadPostponeBucketAssigner(
FileStoreTable table,
long snapshotId,
int defaultParallelism,
@@ -142,15 +143,6 @@ public class PostponeUtils {
!table.coreOptions().postponeTargetRowNumPerBucket().isPresent()
? Collections.emptyMap()
: getPostponeRowCounts(table, snapshotId,
partitionFilter);
- return createPostponeBucketAssigner(
- table, knownNumBuckets, postponeRowCounts, defaultParallelism);
- }
-
- private static PostponeBucketAssigner createPostponeBucketAssigner(
- FileStoreTable table,
- Map<BinaryRow, Integer> knownNumBuckets,
- Map<BinaryRow, Long> postponeRowCounts,
- int defaultParallelism) {
Long targetRowNumPerBucket =
table.coreOptions().postponeTargetRowNumPerBucket().orElse(null);
int defaultBucketNum =
@@ -163,18 +155,36 @@ public class PostponeUtils {
knownNumBuckets, targetRowNumPerBucket, postponeRowCounts,
defaultBucketNum);
}
+ /** Creates snapshot-bound routing metadata. */
public static PostponeBucketRouter createPostponeBucketRouter(
FileStoreTable table,
long snapshotId,
int defaultParallelism,
@Nullable PartitionPredicate partitionFilter) {
- return createPostponeBucketRouter(
+ return newPostponeBucketRouter(
+ table,
+ loadPostponeBucketAssigner(table, snapshotId,
defaultParallelism, partitionFilter));
+ }
+
+ /** 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.");
+ Map<BinaryRow, Integer> copied = new HashMap<>();
+ for (Map.Entry<BinaryRow, Integer> entry :
numBucketsByPartition.entrySet()) {
+ checkArgument(
+ entry.getValue() != null && entry.getValue() > 0,
+ "Postpone bucket number must be positive.");
+ copied.put(entry.getKey().copy(), entry.getValue());
+ }
+ return newPostponeBucketRouter(
table,
- createPostponeBucketAssigner(
- table, snapshotId, defaultParallelism,
partitionFilter));
+ new PostponeBucketAssigner(copied, null,
Collections.emptyMap(), defaultBucketNum));
}
- private static PostponeBucketRouter createPostponeBucketRouter(
+ private static PostponeBucketRouter newPostponeBucketRouter(
FileStoreTable table, PostponeBucketAssigner bucketAssigner) {
List<String> trimmedPrimaryKeys = table.schema().trimmedPrimaryKeys();
int[] bucketKeyMapping =
@@ -249,6 +259,93 @@ public class PostponeUtils {
}
}
+ /**
+ * Decides the target bucket number from an exactly measured staged batch.
+ *
+ * <p>Previously committed postpone files are intentionally excluded. An
existing layout is
+ * retained while the staged batch's required bucket number stays within
the configured rescale
+ * load factor. This avoids paying for a layout rewrite for ordinary size
differences while
+ * protecting a large batch from being funneled into too few writers.
+ */
+ public static FixedBucketDecision decideFixedBucketNum(
+ long stagedRowCount,
+ long stagedFileSize,
+ @Nullable Integer existingBucketNum,
+ CoreOptions options) {
+ checkArgument(stagedRowCount >= 0, "Staged row count cannot be
negative.");
+ checkArgument(stagedFileSize >= 0, "Staged file size cannot be
negative.");
+ checkArgument(
+ existingBucketNum == null || existingBucketNum > 0,
+ "Existing bucket number must be positive.");
+
+ int maxBucketNum =
options.postponeBatchWriteFixedBucketMaxParallelism();
+ checkArgument(
+ maxBucketNum > 0,
+ "Option '%s' must be greater than 0.",
+
CoreOptions.POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM.key());
+ int rescaleLoadFactor =
options.postponeBatchWriteFixedBucketRescaleLoadFactor();
+ checkArgument(
+ rescaleLoadFactor > 0,
+ "Option '%s' must be greater than 0.",
+
CoreOptions.POSTPONE_BATCH_WRITE_FIXED_BUCKET_RESCALE_LOAD_FACTOR.key());
+
+ BigInteger requiredBucketNum;
+ if (options.postponeTargetRowNumPerBucket().isPresent()) {
+ long targetRowCount =
options.postponeTargetRowNumPerBucket().get();
+ checkArgument(
+ targetRowCount > 0,
+ "Option '%s' must be greater than 0.",
+ CoreOptions.POSTPONE_TARGET_ROW_NUM_PER_BUCKET.key());
+ requiredBucketNum =
+ ceilDiv(BigInteger.valueOf(stagedRowCount),
BigInteger.valueOf(targetRowCount));
+ } else {
+ long targetSize = options.postponeTargetSizePerBucket();
+ checkArgument(
+ targetSize > 0,
+ "Option '%s' must be greater than 0.",
+ CoreOptions.POSTPONE_TARGET_SIZE_PER_BUCKET.key());
+ requiredBucketNum =
+ ceilDiv(BigInteger.valueOf(stagedFileSize),
BigInteger.valueOf(targetSize));
+ }
+
+ requiredBucketNum = requiredBucketNum.max(BigInteger.ONE);
+ int suggestedBucketNum = roundUpToPowerOfTwo(requiredBucketNum,
maxBucketNum);
+ boolean requiresRescale =
+ existingBucketNum != null
+ && requiredBucketNum.compareTo(
+ BigInteger.valueOf(existingBucketNum)
+
.multiply(BigInteger.valueOf(rescaleLoadFactor)))
+ > 0
+ && suggestedBucketNum > existingBucketNum;
+ int targetBucketNum;
+ if (existingBucketNum == null) {
+ targetBucketNum = suggestedBucketNum;
+ } else if (requiresRescale) {
+ targetBucketNum = suggestedBucketNum;
+ } else {
+ targetBucketNum = existingBucketNum;
+ }
+ return new FixedBucketDecision(targetBucketNum, requiresRescale);
+ }
+
+ private static BigInteger ceilDiv(BigInteger value, BigInteger divisor) {
+ checkArgument(divisor.signum() > 0, "Bucket target must be positive.");
+ if (value.signum() <= 0) {
+ return BigInteger.ZERO;
+ }
+ return
value.subtract(BigInteger.ONE).divide(divisor).add(BigInteger.ONE);
+ }
+
+ private static int roundUpToPowerOfTwo(BigInteger value, int upperBound) {
+ int cappedValue =
value.min(BigInteger.valueOf(upperBound)).intValueExact();
+ if (cappedValue <= 1) {
+ return 1;
+ }
+
+ long roundedValue = (long) Integer.highestOneBit(cappedValue - 1) << 1;
+ return (int) Math.min(roundedValue, upperBound);
+ }
+
public static Map<BinaryRow, Integer> getKnownNumBuckets(FileStoreTable
table) {
return getKnownNumBuckets(
table.store().newScan().onlyReadRealBuckets().readSimpleEntries());
@@ -256,7 +353,22 @@ public class PostponeUtils {
public static Map<BinaryRow, Integer> getKnownNumBuckets(
FileStoreTable table, long snapshotId) {
- return getKnownNumBuckets(table, snapshotId, null);
+ return getKnownNumBuckets(table, snapshotId, (PartitionPredicate)
null);
+ }
+
+ /** Returns known real-bucket counts only for the specified partitions. */
+ public static Map<BinaryRow, Integer> getKnownNumBuckets(
+ FileStoreTable table, long snapshotId, List<BinaryRow> partitions)
{
+ if (partitions.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ return getKnownNumBuckets(
+ table.store()
+ .newScan()
+ .withSnapshot(snapshotId)
+ .withPartitionFilter(partitions)
+ .onlyReadRealBuckets()
+ .readSimpleEntries());
}
static Map<BinaryRow, Integer> getKnownNumBuckets(
@@ -386,6 +498,26 @@ public class PostponeUtils {
}
}
+ /** Bucket decision for an exactly measured staged batch. */
+ public static final class FixedBucketDecision {
+
+ private final int targetBucketNum;
+ private final boolean requiresRescale;
+
+ private FixedBucketDecision(int targetBucketNum, boolean
requiresRescale) {
+ this.targetBucketNum = targetBucketNum;
+ this.requiresRescale = requiresRescale;
+ }
+
+ public int targetBucketNum() {
+ return targetBucketNum;
+ }
+
+ public boolean requiresRescale() {
+ return requiresRescale;
+ }
+ }
+
/** Snapshot-bound routing metadata for postpone records. */
public static final class PostponeBucketRouter implements Serializable {
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 75eda5624e..c0561c09db 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
@@ -56,7 +56,7 @@ public final class PostponeMergeReadBuilder implements
Serializable {
private static final long serialVersionUID = 1L;
private final FileStoreTable table;
- private final Snapshot snapshot;
+ private final @Nullable Snapshot snapshot;
@Nullable private Predicate filter;
@Nullable private PartitionPredicate partitionFilter;
@@ -65,11 +65,19 @@ public final class PostponeMergeReadBuilder implements
Serializable {
@Nullable private transient String readProtectionTagName;
private int defaultBucketNum = 1;
- private PostponeMergeReadBuilder(FileStoreTable table, Snapshot snapshot) {
+ private PostponeMergeReadBuilder(FileStoreTable table, @Nullable Snapshot
snapshot) {
this.table = table;
this.snapshot = snapshot;
}
+ /** Creates a builder for execution-engine supplied postpone and
real-bucket splits. */
+ public static PostponeMergeReadBuilder createForSplits(FileStoreTable
table) {
+ checkArgument(
+ table.bucketMode() == BucketMode.POSTPONE_MODE &&
!table.primaryKeys().isEmpty(),
+ "Postpone merge read requires a primary-key postpone bucket
table.");
+ return new PostponeMergeReadBuilder(table, null);
+ }
+
/** Creates a snapshot-bound builder when the selected partitions contain
postpone files. */
public static Optional<PostponeMergeReadBuilder> create(
FileStoreTable table, @Nullable PartitionPredicate
partitionFilter) {
@@ -179,6 +187,7 @@ public final class PostponeMergeReadBuilder implements
Serializable {
}
public PostponeMergePlan plan() {
+ checkArgument(snapshot != null, "Snapshot-bound postpone merge plan
requires a snapshot.");
RowType resultReadType = resultReadType();
RowType mergeReadType = mergeReadType(resultReadType);
@@ -222,6 +231,21 @@ public final class PostponeMergeReadBuilder implements
Serializable {
return plan;
}
+ /** Builds a plan from splits and routing metadata supplied by an
execution engine. */
+ public PostponeMergePlan plan(
+ List<DataSplit> realSplits,
+ List<DataSplit> postponeSplits,
+ PostponeUtils.PostponeBucketRouter bucketRouter) {
+ RowType resultReadType = resultReadType();
+ return new PostponeMergePlan(
+ realSplits,
+ PostponeUtils.groupPostponeFiles(postponeSplits),
+ bucketRouter,
+ keyType(),
+ resultReadType,
+ 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.");
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 40b36bfb96..1d07359b6d 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
@@ -18,6 +18,7 @@
package org.apache.paimon.table;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.FileStore;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
@@ -72,6 +73,28 @@ public class PostponeUtilsTest {
verify(scan).withPartitionFilter(partitionFilter);
}
+ @Test
+ public void testGetKnownNumBucketsByPartitions() {
+ BinaryRow partition = partition(1);
+ List<BinaryRow> partitions = Collections.singletonList(partition);
+ SimpleFileEntry entry = mock(SimpleFileEntry.class);
+ when(entry.partition()).thenReturn(partition);
+ when(entry.totalBuckets()).thenReturn(4);
+
+ FileStoreScan scan = mock(FileStoreScan.class, RETURNS_SELF);
+
when(scan.readSimpleEntries()).thenReturn(Collections.singletonList(entry));
+ FileStore store = mock(FileStore.class);
+ when(store.newScan()).thenReturn(scan);
+ FileStoreTable table = mock(FileStoreTable.class);
+ when(table.store()).thenReturn(store);
+
+ assertThat(PostponeUtils.getKnownNumBuckets(table, 5L, partitions))
+ .containsEntry(partition, 4);
+ verify(scan).withSnapshot(5L);
+ verify(scan).onlyReadRealBuckets();
+ verify(scan).withPartitionFilter(partitions);
+ }
+
@Test
public void testGetPostponeRowCountsFromSnapshot() {
BinaryRow partition = partition(1);
@@ -268,6 +291,84 @@ public class PostponeUtilsTest {
.isEqualTo(7);
}
+ @Test
+ public void testDecideFixedBucketNum() {
+ Map<String, String> optionMap = new HashMap<>();
+ optionMap.put(CoreOptions.POSTPONE_TARGET_ROW_NUM_PER_BUCKET.key(),
"1");
+
optionMap.put(CoreOptions.POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM.key(),
"16");
+ optionMap.put(
+
CoreOptions.POSTPONE_BATCH_WRITE_FIXED_BUCKET_RESCALE_LOAD_FACTOR.key(), "32");
+ CoreOptions options = CoreOptions.fromMap(optionMap);
+
+ PostponeUtils.FixedBucketDecision rounded =
+ PostponeUtils.decideFixedBucketNum(3, 0, null, options);
+ assertThat(rounded.targetBucketNum()).isEqualTo(4);
+ assertThat(rounded.requiresRescale()).isFalse();
+
+ PostponeUtils.FixedBucketDecision capped =
+ PostponeUtils.decideFixedBucketNum((long) Integer.MAX_VALUE +
1, 0, null, options);
+ assertThat(capped.targetBucketNum()).isEqualTo(16);
+ assertThat(capped.requiresRescale()).isFalse();
+
+ PostponeUtils.FixedBucketDecision atLoadFactor =
+ PostponeUtils.decideFixedBucketNum(224, 0, 7, options);
+ assertThat(atLoadFactor.targetBucketNum()).isEqualTo(7);
+ assertThat(atLoadFactor.requiresRescale()).isFalse();
+
+ PostponeUtils.FixedBucketDecision aboveLoadFactor =
+ PostponeUtils.decideFixedBucketNum(225, 0, 7, options);
+ assertThat(aboveLoadFactor.targetBucketNum()).isEqualTo(16);
+ assertThat(aboveLoadFactor.requiresRescale()).isTrue();
+
+ Map<String, String> cappedOptionMap = new HashMap<>(optionMap);
+ cappedOptionMap.put(
+
CoreOptions.POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM.key(), "4");
+ PostponeUtils.FixedBucketDecision cappedBelowExisting =
+ PostponeUtils.decideFixedBucketNum(225, 0, 7,
CoreOptions.fromMap(cappedOptionMap));
+ assertThat(cappedBelowExisting.targetBucketNum()).isEqualTo(7);
+ assertThat(cappedBelowExisting.requiresRescale()).isFalse();
+
+ PostponeUtils.FixedBucketDecision largerExisting =
+ PostponeUtils.decideFixedBucketNum(257, 0, 8, options);
+ assertThat(largerExisting.targetBucketNum()).isEqualTo(16);
+ assertThat(largerExisting.requiresRescale()).isTrue();
+
+ Map<String, String> lowerLoadFactorOptions = new HashMap<>(optionMap);
+ lowerLoadFactorOptions.put(
+
CoreOptions.POSTPONE_BATCH_WRITE_FIXED_BUCKET_RESCALE_LOAD_FACTOR.key(), "4");
+ PostponeUtils.FixedBucketDecision lowerLoadFactor =
+ PostponeUtils.decideFixedBucketNum(
+ 29, 0, 7, CoreOptions.fromMap(lowerLoadFactorOptions));
+ assertThat(lowerLoadFactor.targetBucketNum()).isEqualTo(16);
+ assertThat(lowerLoadFactor.requiresRescale()).isTrue();
+ }
+
+ @Test
+ public void testDecideFixedBucketNumRejectsInvalidRescaleLoadFactor() {
+ Map<String, String> optionMap = new HashMap<>();
+ optionMap.put(CoreOptions.POSTPONE_TARGET_ROW_NUM_PER_BUCKET.key(),
"1");
+
optionMap.put(CoreOptions.POSTPONE_BATCH_WRITE_FIXED_BUCKET_RESCALE_LOAD_FACTOR.key(),
"0");
+
+ assertThatThrownBy(
+ () ->
+ PostponeUtils.decideFixedBucketNum(
+ 1, 0, 1,
CoreOptions.fromMap(optionMap)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+
CoreOptions.POSTPONE_BATCH_WRITE_FIXED_BUCKET_RESCALE_LOAD_FACTOR.key());
+ }
+
+ @Test
+ public void testDecideFixedBucketNumByStagedFileSize() {
+ Map<String, String> optionMap = new HashMap<>();
+ optionMap.put(CoreOptions.POSTPONE_TARGET_SIZE_PER_BUCKET.key(), "100
b");
+
optionMap.put(CoreOptions.POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM.key(),
"32");
+
+ PostponeUtils.FixedBucketDecision decision =
+ PostponeUtils.decideFixedBucketNum(10, 1000, null,
CoreOptions.fromMap(optionMap));
+ assertThat(decision.targetBucketNum()).isEqualTo(16);
+ }
+
private static BinaryRow partition(int value) {
BinaryRow row = new BinaryRow(1);
BinaryRowWriter writer = new BinaryRowWriter(row);
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 d27f6dd615..c3a020cbd3 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
@@ -1007,6 +1007,43 @@ public class PostponeBucketTableITCase extends
AbstractTestBase {
"+I[5, 53]");
}
+ @Test
+ public void testFixedBucketWriteDoesNotCompact() throws Exception {
+ String warehouse = getTempDirPath();
+ TableEnvironment tEnv =
+ tableEnvironmentBuilder()
+ .batchMode()
+ .setConf(TableConfigOptions.TABLE_DML_SYNC, true)
+ .build();
+
+ tEnv.executeSql(
+ "CREATE CATALOG mycat WITH (\n"
+ + " 'type' = 'paimon',\n"
+ + " 'warehouse' = '"
+ + warehouse
+ + "'\n"
+ + ")");
+ tEnv.executeSql("USE CATALOG mycat");
+ tEnv.executeSql(
+ "CREATE TABLE T (\n"
+ + " k INT,\n"
+ + " v STRING,\n"
+ + " PRIMARY KEY (k) NOT ENFORCED\n"
+ + ") WITH (\n"
+ + " 'bucket' = '-2',\n"
+ + "
'postpone.batch-write-fixed-bucket.max-parallelism' = '1',\n"
+ + " 'num-sorted-run.compaction-trigger' = '2'\n"
+ + ")");
+
+ tEnv.executeSql("INSERT INTO T VALUES (1, 'a')").await();
+ tEnv.executeSql("INSERT INTO T VALUES (2, 'b')").await();
+
+ assertThat(collect(tEnv.executeSql("SELECT * FROM T")))
+ .containsExactlyInAnyOrder("+I[1, a]", "+I[2, b]");
+ assertThat(collect(tEnv.executeSql("SELECT * FROM `T$files` WHERE
level = 0"))).hasSize(2);
+ assertThat(collect(tEnv.executeSql("SELECT * FROM `T$files` WHERE
level > 0"))).isEmpty();
+ }
+
@Test
public void testWriteFixedBucketWithDifferentBucketNumber() throws
Exception {
String warehouse = getTempDirPath();
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
index 66f7f31615..36ab181aec 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
@@ -187,7 +187,7 @@ public class CompactChainTableProcedure extends
BaseProcedure {
if (partitionExists) {
Map<String, String> staticPartition =
SparkProcedureUtils.parseStaticPartition(spark(),
targetPartition);
- writer.writeBuilder().withOverwrite(staticPartition);
+ writer.withOverwrite(staticPartition);
}
writer.commit(writer.write(datasetForWrite));
LOG.info("Successfully compacted partition {} to snapshot branch.",
partitionStr);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
index 27dd435aba..1e156ca42a 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
@@ -671,7 +671,7 @@ public class CompactProcedure extends BaseProcedure {
if (datasetForWrite != null) {
PaimonSparkWriter writer = PaimonSparkWriter.apply(table);
// Use dynamic partition overwrite
- writer.writeBuilder().withOverwrite();
+ writer.withOverwrite();
writer.commit(writer.write(datasetForWrite));
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java
index 404a78db78..232f4370b5 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java
@@ -186,7 +186,7 @@ public class RescaleProcedure extends BaseProcedure {
FileStoreTable rescaledTable =
table.copy(table.schema().copy(bucketOptions));
PaimonSparkWriter writer = PaimonSparkWriter.apply(rescaledTable);
- writer.writeBuilder().withOverwrite();
+ writer.withOverwrite();
writer.commit(writer.write(datasetForRead));
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeInputScan.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeInputScan.scala
index f0d5cdb1fe..9721320c6d 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeInputScan.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeInputScan.scala
@@ -255,7 +255,7 @@ private[spark] object PostponeMergeInputScan {
private def bucketKey(split: DataSplit) = (split.partition(), split.bucket())
- private def mergeRealSplits(splits: Seq[DataSplit]): DataSplit = {
+ private[spark] def mergeRealSplits(splits: Seq[DataSplit]): DataSplit = {
if (splits.size == 1) {
splits.head
} else {
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
new file mode 100644
index 0000000000..1ba596b4fe
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala
@@ -0,0 +1,717 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark
+
+import org.apache.paimon.{CoreOptions, KeyValue, Snapshot}
+import org.apache.paimon.data.{BinaryRow, InternalRow}
+import org.apache.paimon.data.serializer.InternalRowSerializer
+import org.apache.paimon.operation.FileSystemWriteRestore
+import org.apache.paimon.options.Options
+import org.apache.paimon.postpone.BucketFiles
+import org.apache.paimon.reader.RecordReaderIterator
+import org.apache.paimon.table.{BucketMode, FileStoreTable, PostponeUtils}
+import org.apache.paimon.table.PostponeUtils.PostponeBucketRouter
+import org.apache.paimon.table.sink.{CommitMessage, CommitMessageImpl,
CommitMessageSerializer, PostponeFixedBucketWriteBuilder, TableWriteImpl}
+import org.apache.paimon.table.source.{DataSplit, PostponeMergePlan,
PostponeMergeReadBuilder, SplitSerializer}
+import org.apache.paimon.types.RowKind
+import org.apache.paimon.utils.{IteratorRecordReader, SerializationUtils}
+
+import org.apache.spark.{Partitioner, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.{PaimonUtils, SparkSession}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+/**
+ * Completes a fixed-bucket batch write from uncommitted postpone files.
+ *
+ * Job 1 has already materialized the source rows into bucket -2. This
coordinator derives the
+ * actual target partitions and bucket counts from those files. When necessary
it first rescales
+ * existing real-bucket data, then writes and commits the current batch to
real buckets. Committed
+ * postpone files are left to the regular postpone compaction path.
+ */
+private[spark] class SparkPostponeStagedCommitter(
+ table: FileStoreTable,
+ @transient spark: SparkSession,
+ baseSnapshotId: Option[Long],
+ overwritePartitionSpec: Option[Map[String, String]])
+ extends Serializable {
+
+ import SparkPostponeStagedCommitter._
+
+ private val coreOptions = table.coreOptions()
+ private val fixedWriteCommitUser = coreOptions.createCommitUser()
+ private val fixedWriteTable = {
+ val options = new java.util.HashMap[String, String]()
+ options.put(
+ CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(),
+ baseSnapshotId.getOrElse(0L).toString)
+ // Every overwrite in this coordinator supplies its exact BinaryRow
partitions. Disable the
+ // table-level dynamic rewrite so an empty rewritten partition is still
removed, and implement
+ // INSERT OVERWRITE dynamic-partition semantics explicitly in
commitCurrentBatch.
+ options.put(CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key(), "false")
+ table.copy(options)
+ }
+ private val fixedWriteBuilder =
fixedWriteTable.newPostponeFixedBucketWriteBuilder()
+ private val rescaleWriteBuilder = {
+ val options = new java.util.HashMap[String, String]()
+ options.put(CoreOptions.CHANGELOG_PRODUCER.key(),
CoreOptions.ChangelogProducer.NONE.toString)
+ fixedWriteTable.copy(options).newPostponeFixedBucketWriteBuilder()
+ }
+ private val ignoreEmptyCommit = new Options(table.options())
+ .getOptional(CoreOptions.SNAPSHOT_IGNORE_EMPTY_COMMIT)
+ .orElse(true)
+
+ def commit(
+ stagedMessages: Seq[CommitMessage],
+ operation: Snapshot.Operation): Seq[CommitMessage] = {
+ val (stageMessages, passThroughMessages) =
stagedMessages.partition(isStageMessage)
+ var cleanupStageMessages = stageMessages
+ var pendingRescaleMessages = Seq.empty[CommitMessage]
+ var pendingFinalMessages = passThroughMessages
+ var rescaleCommitStarted = false
+ var finalCommitStarted = false
+
+ try {
+ val normalizedStageMessages = mergeBucketMessages(stageMessages)
+ cleanupStageMessages = normalizedStageMessages
+ val stagedSplits = splitsFromMessages(normalizedStageMessages)
+ val stats = aggregateStats(stagedSplits)
+ val touchedPartitions = stats.keysIterator.map(_.copy()).toSeq
+ if (touchedPartitions.isEmpty) {
+ if (
+ overwritePartitionSpec.isDefined || passThroughMessages.nonEmpty ||
!ignoreEmptyCommit
+ ) {
+ finalCommitStarted = true
+ commitCurrentBatch(passThroughMessages, Seq.empty, operation)
+ }
+ return passThroughMessages
+ }
+
+ val existingBuckets: Map[BinaryRow, Int] = baseSnapshotId
+ .map(
+ id =>
+ PostponeUtils
+ .getKnownNumBuckets(table, id, touchedPartitions.asJava)
+ .asScala
+ .iterator
+ .map { case (partition, buckets) => partition ->
buckets.intValue() }
+ .toMap)
+ .getOrElse(Map.empty[BinaryRow, Int])
+ val decisions = touchedPartitions.map {
+ partition =>
+ val stage = stats(partition)
+ val decision = PostponeUtils.decideFixedBucketNum(
+ stage.rowCount,
+ stage.fileSize,
+ existingBuckets.get(partition).map(Int.box).orNull,
+ coreOptions
+ )
+ partition -> decision
+ }.toMap
+
+ val targetBuckets = decisions.map {
+ case (partition, decision) => partition -> decision.targetBucketNum()
+ }
+
+ if (overwritePartitionSpec.isDefined) {
+ val rewrittenMessages =
+ writePostponeRecords(
+ stagedSplits,
+ targetBuckets,
+ replacePreviousFiles = true,
+ restoreSnapshotId = None)
+ pendingFinalMessages = rewrittenMessages ++ passThroughMessages
+ finalCommitStarted = true
+ commitCurrentBatch(pendingFinalMessages, touchedPartitions, operation)
+ return pendingFinalMessages
+ }
+
+ val rescaleBucketNums = decisions.collect {
+ case (partition, decision) if decision.requiresRescale() =>
+ partition -> decision.targetBucketNum()
+ }
+ if (rescaleBucketNums.nonEmpty) {
+ pendingRescaleMessages =
+ rewriteRealBuckets(readRealSplits(rescaleBucketNums.keys.toSeq),
rescaleBucketNums)
+ rescaleCommitStarted = true
+ commitRescale(pendingRescaleMessages, rescaleBucketNums)
+ pendingRescaleMessages = Seq.empty
+ }
+
+ // After rescale, the current batch must restore from the new snapshot.
Otherwise it restores
+ // from the snapshot captured before the staging job.
+ val currentWriteSnapshotId =
+ if (rescaleBucketNums.nonEmpty) {
+ Option(table.snapshotManager().latestSnapshot()).map(_.id())
+ } else {
+ baseSnapshotId
+ }
+ val currentMessages =
+ writePostponeRecords(
+ stagedSplits,
+ targetBuckets,
+ replacePreviousFiles = false,
+ restoreSnapshotId = currentWriteSnapshotId)
+ pendingFinalMessages = currentMessages ++ passThroughMessages
+ finalCommitStarted = true
+ commitCurrentBatch(pendingFinalMessages, touchedPartitions, operation)
+ pendingFinalMessages
+ } catch {
+ case error: Throwable =>
+ if (!rescaleCommitStarted && pendingRescaleMessages.nonEmpty) {
+ abortMessages(pendingRescaleMessages)
+ }
+ if (!finalCommitStarted && pendingFinalMessages.nonEmpty) {
+ abortMessages(pendingFinalMessages)
+ }
+ throw error
+ } finally {
+ // Staged files were never visible, so they are always safe to delete,
including when the
+ // final commit returned an unknown result.
+ abortMessages(cleanupStageMessages)
+ }
+ }
+
+ private def aggregateStats(stagedSplits: Seq[DataSplit]): Map[BinaryRow,
PartitionStats] = {
+ val result = mutable.HashMap.empty[BinaryRow, PartitionStats]
+ stagedSplits.foreach {
+ split =>
+ val previous = result.getOrElse(split.partition(), PartitionStats(0L,
0L))
+ val splitFileSize = split
+ .dataFiles()
+ .asScala
+ .iterator
+ .map(_.fileSize())
+ .foldLeft(0L)(Math.addExact)
+ result.put(
+ split.partition().copy(),
+ PartitionStats(
+ Math.addExact(previous.rowCount, split.rowCount()),
+ Math.addExact(previous.fileSize, splitFileSize)))
+ }
+ result.toMap
+ }
+
+ private def splitsFromMessages(messages: Seq[CommitMessage]): Seq[DataSplit]
= {
+ messages.collect {
+ case message: CommitMessageImpl if message.bucket() ==
BucketMode.POSTPONE_BUCKET =>
+ val files = message.newFilesIncrement().newFiles().asScala ++
+ message.compactIncrement().compactAfter().asScala
+ if (files.isEmpty) {
+ None
+ } else {
+ Some(
+ DataSplit
+ .builder()
+ .withSnapshot(baseSnapshotId.getOrElse(0L))
+ .withPartition(message.partition())
+ .withBucket(BucketMode.POSTPONE_BUCKET)
+ .withBucketPath(
+ table
+ .store()
+ .pathFactory()
+ .bucketPath(message.partition(), BucketMode.POSTPONE_BUCKET)
+ .toString)
+ .withTotalBuckets(message.totalBuckets())
+ .withDataFiles(files.asJava)
+ .isStreaming(false)
+ .rawConvertible(false)
+ .build())
+ }
+ }.flatten
+ }
+
+ private def isStageMessage(message: CommitMessage): Boolean = message match {
+ case commit: CommitMessageImpl => commit.bucket() ==
BucketMode.POSTPONE_BUCKET
+ case _ => false
+ }
+
+ private def readRealSplits(partitions: Seq[BinaryRow]): Seq[DataSplit] = {
+ baseSnapshotId.toSeq.flatMap {
+ snapshotId =>
+ table
+ .newSnapshotReader()
+ .withSnapshot(snapshotId)
+ .withPartitionFilter(partitions.asJava)
+ .onlyReadRealBuckets()
+ .read()
+ .dataSplits()
+ .asScala
+ }
+ }
+
+ private def writePostponeRecords(
+ postponeSplits: Seq[DataSplit],
+ bucketNums: Map[BinaryRow, Int],
+ replacePreviousFiles: Boolean,
+ restoreSnapshotId: Option[Long]): Seq[CommitMessage] = {
+ val router = createRouter(bucketNums)
+ val readBuilder = PostponeMergeReadBuilder.createForSplits(table)
+ val corePlan = readBuilder.plan(Seq.empty[DataSplit].asJava,
postponeSplits.asJava, router)
+ val records = routePostponeRecords(readBuilder, corePlan)
+ val partitioned = records.repartitionAndSortWithinPartitions(
+ new BucketGroupPartitioner(shuffleParallelism(bucketNums)))
+ writeRoutedRecords(
+ partitioned,
+ bucketNums,
+ replacePreviousFiles,
+ restoreSnapshotId = restoreSnapshotId,
+ writeBuilder = fixedWriteBuilder)
+ }
+
+ private def rewriteRealBuckets(
+ realSplits: Seq[DataSplit],
+ targetBucketNums: Map[BinaryRow, Int]): Seq[CommitMessage] = {
+ val targetRouter = createRouter(targetBucketNums)
+ val readBuilder = PostponeMergeReadBuilder.createForSplits(table)
+ val corePlan = readBuilder.plan(realSplits.asJava,
Seq.empty[DataSplit].asJava, targetRouter)
+ val routedRecords = routeRealRecords(readBuilder, corePlan)
+ .repartitionAndSortWithinPartitions(
+ new BucketGroupPartitioner(shuffleParallelism(targetBucketNums)))
+ writeRoutedRecords(
+ routedRecords,
+ targetBucketNums,
+ replacePreviousFiles = true,
+ restoreSnapshotId = None,
+ writeBuilder = rescaleWriteBuilder)
+ }
+
+ private def writeRoutedRecords(
+ records: RDD[(BucketOrderKey, PostponeRecord)],
+ bucketNums: Map[BinaryRow, Int],
+ replacePreviousFiles: Boolean,
+ restoreSnapshotId: Option[Long],
+ writeBuilder: PostponeFixedBucketWriteBuilder): Seq[CommitMessage] = {
+ val written = records.mapPartitions {
+ input =>
+ if (!input.hasNext) {
+ Iterator.empty
+ } else {
+ val ioManager = SparkUtils.createIOManager()
+ val write = writeBuilder
+ .newWrite(fixedWriteCommitUser, null)
+ .withIOManager(ioManager)
+ .asInstanceOf[TableWriteImpl[InternalRow]]
+ if (replacePreviousFiles) {
+ write.withIgnorePreviousFiles(true)
+ } else {
+ restoreSnapshotId.foreach {
+ id =>
+ write.withWriteRestore(
+ new FileSystemWriteRestore(
+ table.coreOptions(),
+ table.snapshotManager(),
+ table.store().newScan(),
+ table.store().newIndexFileHandler(),
+ id))
+ }
+ }
+ try {
+ val buffered = input.buffered
+ while (buffered.hasNext) {
+ val first = buffered.head._1
+ val partition = deserializePartition(first.partition)
+ val bucket = first.bucket
+ while (buffered.hasNext && buffered.head._1.sameBucket(first)) {
+ val record = buffered.next()._2
+ val row = SerializationUtils.deserializeBinaryRow(record.value)
+ row.setRowKind(RowKind.fromByteValue(record.rowKind))
+ write.writeAndReturn(row, bucket, bucketNums(partition))
+ }
+ }
+ val serializer = new CommitMessageSerializer()
+ val commitMessages = write.prepareCommit().asScala.toVector
+ reportOutputMetrics(commitMessages)
+ Iterator.single(
+ commitMessages
+ .map(serializer.serialize))
+ } finally {
+ try {
+ write.close()
+ } finally {
+ ioManager.close()
+ }
+ }
+ }
+ }
+ deserializeCommitMessages(written.collect().iterator.flatten.toVector)
+ }
+
+ private def deserializeCommitMessages(
+ serializedMessages: Iterable[Array[Byte]]): Seq[CommitMessage] = {
+ val serializer = new CommitMessageSerializer()
+ serializedMessages.iterator
+ .map(serializer.deserialize(serializer.getVersion, _))
+ .toVector
+ }
+
+ private def routeRealRecords(
+ readBuilder: PostponeMergeReadBuilder,
+ plan: PostponeMergePlan): RDD[(BucketOrderKey, PostponeRecord)] = {
+ val splits = plan.realSplits().asScala.toSeq
+ if (splits.isEmpty) {
+ return spark.sparkContext.emptyRDD
+ }
+ val keyType = plan.keyType()
+ val resultReadType = plan.resultReadType()
+ val router = plan.bucketRouter()
+ val indexedSplits = splits
+ .groupBy(split => (split.partition(), split.bucket()))
+ .values
+ .map(bucketSplits =>
PostponeMergeInputScan.mergeRealSplits(bucketSplits.toSeq))
+ .toSeq
+ .map(serializeSplit)
+ .zipWithIndex
+ spark.sparkContext
+ .parallelize(
+ indexedSplits,
+ Math.max(1, Math.min(indexedSplits.size,
spark.sparkContext.defaultParallelism)))
+ .mapPartitions {
+ inputs =>
+ val ioManager = SparkUtils.createIOManager()
+ val read = readBuilder.newRead().withIOManager(ioManager)
+ val openReaders =
mutable.HashSet.empty[RecordReaderIterator[InternalRow]]
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit] {
+ _ =>
+ try {
+ openReaders.foreach(_.close())
+ } finally {
+ ioManager.close()
+ }
+ })
+ val keyExtractor = table.createRowKeyExtractor()
+ val keySerializer = new InternalRowSerializer(keyType)
+ val valueSerializer = new InternalRowSerializer(resultReadType)
+ inputs.flatMap {
+ case (serializedSplit, splitOrder) =>
+ val split = deserializeSplit(serializedSplit)
+ val splitPartition = split.partition()
+ val partitionBytes =
serializePartition(splitPartition).toIndexedSeq
+ val emptyRecords =
+ new IteratorRecordReader[KeyValue](
+ java.util.Collections.emptyList[KeyValue]().iterator())
+ val reader = new RecordReaderIterator[InternalRow](
+ read.createBucketMergeReader(split, emptyRecords))
+ mapReader(reader, openReaders) {
+ (row, localOrder) =>
+ keyExtractor.setRecord(row)
+ val key =
keySerializer.toBinaryRow(keyExtractor.trimmedPrimaryKey())
+ val bucket = router.bucket(splitPartition, key)
+ BucketOrderKey(
+ partitionBytes,
+ bucket,
+ splitOrder.toLong,
+ localOrder) -> PostponeRecord(
+ row.getRowKind.toByteValue,
+
SerializationUtils.serializeBinaryRow(valueSerializer.toBinaryRow(row))
+ )
+ }
+ }
+ }
+ }
+
+ private def routePostponeRecords(
+ readBuilder: PostponeMergeReadBuilder,
+ plan: PostponeMergePlan): RDD[(BucketOrderKey, PostponeRecord)] = {
+ val splits = plan.postponeSplits().asScala.toSeq
+ if (splits.isEmpty) {
+ return spark.sparkContext.emptyRDD
+ }
+ val keyType = plan.keyType()
+ val mergeReadType = plan.mergeReadType()
+ val router = plan.bucketRouter()
+ val indexedSplits = splits.map(serializeSplit).zipWithIndex
+ spark.sparkContext
+ .parallelize(
+ indexedSplits,
+ Math.max(1, Math.min(indexedSplits.size,
spark.sparkContext.defaultParallelism)))
+ .mapPartitions {
+ inputs =>
+ val ioManager = SparkUtils.createIOManager()
+ val read = readBuilder.newRead().withIOManager(ioManager)
+ val openReaders =
mutable.HashSet.empty[RecordReaderIterator[KeyValue]]
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit] {
+ _ =>
+ try {
+ openReaders.foreach(_.close())
+ } finally {
+ ioManager.close()
+ }
+ })
+ val keySerializer = new InternalRowSerializer(keyType)
+ val valueSerializer = new InternalRowSerializer(mergeReadType)
+ inputs.flatMap {
+ case (serializedSplit, writerOrder) =>
+ val split = deserializeSplit(serializedSplit)
+ val splitPartition = split.partition()
+ val partitionBytes =
serializePartition(splitPartition).toIndexedSeq
+ val reader = new
RecordReaderIterator[KeyValue](read.createPostponeReader(split))
+ mapReader(reader, openReaders) {
+ (keyValue, localOrder) =>
+ val key = keySerializer.toBinaryRow(keyValue.key())
+ val bucket = router.bucket(splitPartition, key)
+ BucketOrderKey(
+ partitionBytes,
+ bucket,
+ writerOrder.toLong,
+ localOrder) -> PostponeRecord(
+ keyValue.valueKind().toByteValue,
+ SerializationUtils.serializeBinaryRow(
+ valueSerializer.toBinaryRow(keyValue.value()))
+ )
+ }
+ }
+ }
+ }
+
+ private def mergeBucketMessages(messages: Seq[CommitMessage]):
Seq[CommitMessageImpl] = {
+ val buckets = mutable.LinkedHashMap.empty[(BinaryRow, Int), BucketFiles]
+ messages.foreach {
+ case message: CommitMessageImpl =>
+ val key = (message.partition(), message.bucket())
+ val files = buckets.getOrElseUpdate(
+ key,
+ new BucketFiles(
+ table.store().pathFactory().createDataFilePathFactory(key._1,
key._2),
+ table.fileIO()))
+ files.update(message)
+ case other => throw new IllegalArgumentException(s"Unsupported commit
message $other")
+ }
+ buckets.map {
+ case ((partition, bucket), files) =>
+ files.makeMessage(partition, bucket)
+ }.toSeq
+ }
+
+ private def commitRescale(
+ messages: Seq[CommitMessage],
+ rescaleBucketNums: Map[BinaryRow, Int]): Unit = {
+ val snapshotId = baseSnapshotId.getOrElse {
+ throw new IllegalStateException("Cannot rescale real buckets without a
base snapshot.")
+ }
+ // A positive table bucket makes overwrite delete only real buckets.
Commit messages carry the
+ // 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)
+ val commit = commitTable
+ .newCommit(fixedWriteCommitUser)
+ .appendCommitCheckConflict(true)
+ .ignoreEmptyCommit(ignoreEmptyCommit)
+ .withOperation(Snapshot.Operation.OVERWRITE)
+ .withOverwriteStaticPartitions(rescaleBucketNums.keys.toSeq.asJava)
+ try {
+ commit.commit(messages.asJava)
+ } finally {
+ commit.close()
+ }
+ }
+
+ private def commitCurrentBatch(
+ messages: Seq[CommitMessage],
+ touchedPartitions: Seq[BinaryRow],
+ operation: Snapshot.Operation): Unit = {
+ val dynamicPartitionOverwrite =
+ overwritePartitionSpec.exists(_.isEmpty) &&
coreOptions.dynamicPartitionOverwrite()
+ if (!dynamicPartitionOverwrite) {
+ overwritePartitionSpec.foreach(spec =>
fixedWriteBuilder.withOverwrite(spec.asJava))
+ }
+ val commit = fixedWriteBuilder
+ .newCommit(fixedWriteCommitUser, ignoreEmptyCommit)
+ .withOperation(operation)
+ if (dynamicPartitionOverwrite && touchedPartitions.nonEmpty) {
+ commit.withOverwriteStaticPartitions(touchedPartitions.asJava)
+ }
+ try {
+ commit.commit(messages.asJava)
+ } finally {
+ commit.close()
+ }
+ }
+
+ private def abortMessages(messages: Seq[CommitMessage]): Unit = {
+ if (messages.nonEmpty) {
+ try {
+ val commit = table.newBatchWriteBuilder().newCommit()
+ try {
+ commit.abort(messages.asJava)
+ } finally {
+ commit.close()
+ }
+ } catch {
+ case error: Throwable =>
+ // Cleanup failure must not hide the write or commit failure.
Orphan-file cleanup remains
+ // the final safety net for these never-committed files.
+ SparkPostponeStagedCommitter.LOG.warn(
+ s"Failed to clean uncommitted files for table ${table.name()}.",
+ error)
+ }
+ }
+ }
+
+ private def createRouter(bucketNums: Map[BinaryRow, Int]):
PostponeBucketRouter = {
+ val javaBucketNums = new java.util.HashMap[BinaryRow,
Integer](bucketNums.size)
+ bucketNums.foreach {
+ case (partition, buckets) => javaBucketNums.put(partition,
Integer.valueOf(buckets))
+ }
+ PostponeUtils.createPostponeBucketRouter(table, javaBucketNums, 1)
+ }
+
+ private def shuffleParallelism(bucketNums: Map[BinaryRow, Int]): Int = {
+ val useful = bucketNums.values.foldLeft(BigInt(0))(_ + _).max(BigInt(1))
+ useful.min(BigInt(spark.sessionState.conf.numShufflePartitions)).toInt
+ }
+}
+
+private[spark] object SparkPostponeStagedCommitter {
+
+ private val LOG =
org.slf4j.LoggerFactory.getLogger(classOf[SparkPostponeStagedCommitter])
+
+ private case class PartitionStats(rowCount: Long, fileSize: Long)
+
+ private def mapReader[T, R](
+ reader: RecordReaderIterator[T],
+ openReaders: mutable.Set[RecordReaderIterator[T]])(transform: (T, Long)
=> R): Iterator[R] = {
+ openReaders += reader
+ new Iterator[R] {
+ private var nextLocalOrder = 0L
+ private var readerClosed = false
+
+ private def closeReader(): Unit = {
+ if (!readerClosed) {
+ readerClosed = true
+ openReaders -= reader
+ reader.close()
+ }
+ }
+
+ override def hasNext: Boolean = {
+ try {
+ val hasNext = reader.hasNext
+ if (!hasNext) {
+ closeReader()
+ }
+ hasNext
+ } catch {
+ case error: Throwable =>
+ closeReader()
+ throw error
+ }
+ }
+
+ override def next(): R = {
+ if (!hasNext) {
+ throw new NoSuchElementException
+ }
+ try {
+ val result = transform(reader.next(), nextLocalOrder)
+ nextLocalOrder = Math.addExact(nextLocalOrder, 1L)
+ result
+ } catch {
+ case error: Throwable =>
+ closeReader()
+ throw error
+ }
+ }
+ }
+ }
+
+ private case class BucketOrderKey(
+ partition: IndexedSeq[Byte],
+ bucket: Int,
+ writerOrder: Long,
+ localOrder: Long) {
+ def sameBucket(other: BucketOrderKey): Boolean =
+ partition == other.partition && bucket == other.bucket
+ }
+
+ implicit private val bucketOrder: Ordering[BucketOrderKey] = new
Ordering[BucketOrderKey] {
+ override def compare(left: BucketOrderKey, right: BucketOrderKey): Int = {
+ val partitionComparison = compareBytes(left.partition, right.partition)
+ if (partitionComparison != 0) {
+ partitionComparison
+ } else {
+ val bucketComparison = Integer.compare(left.bucket, right.bucket)
+ if (bucketComparison != 0) {
+ bucketComparison
+ } else {
+ val writerComparison = java.lang.Long.compare(left.writerOrder,
right.writerOrder)
+ if (writerComparison != 0) {
+ writerComparison
+ } else {
+ java.lang.Long.compare(left.localOrder, right.localOrder)
+ }
+ }
+ }
+ }
+ }
+
+ private def compareBytes(left: IndexedSeq[Byte], right: IndexedSeq[Byte]):
Int = {
+ val limit = Math.min(left.length, right.length)
+ var index = 0
+ while (index < limit) {
+ val comparison = java.lang.Byte.compare(left(index), right(index))
+ if (comparison != 0) {
+ return comparison
+ }
+ index += 1
+ }
+ Integer.compare(left.length, right.length)
+ }
+
+ private case class BucketGroupPartitioner(override val numPartitions: Int)
extends Partitioner {
+ require(numPartitions > 0, "Shuffle partition number must be positive.")
+
+ override def getPartition(key: Any): Int = {
+ val bucketKey = key.asInstanceOf[BucketOrderKey]
+ Math.floorMod(31 * bucketKey.partition.hashCode() + bucketKey.bucket,
numPartitions)
+ }
+ }
+
+ private case class PostponeRecord(rowKind: Byte, value: Array[Byte])
+
+ private def reportOutputMetrics(messages: Seq[CommitMessage]): Unit = {
+ Option(TaskContext.get()).foreach {
+ taskContext =>
+ val files = messages
+ .collect { case message: CommitMessageImpl => message }
+ .flatMap(_.newFilesIncrement().newFiles().asScala)
+ val bytesWritten =
files.iterator.map(_.fileSize()).foldLeft(0L)(Math.addExact)
+ val recordsWritten =
files.iterator.map(_.rowCount()).foldLeft(0L)(Math.addExact)
+ PaimonUtils.updateOutputMetrics(
+ taskContext.taskMetrics().outputMetrics,
+ bytesWritten,
+ recordsWritten)
+ }
+ }
+
+ private def serializePartition(partition: BinaryRow): Array[Byte] =
+ SerializationUtils.serializeBinaryRow(partition)
+
+ private def deserializePartition(partition: IndexedSeq[Byte]): BinaryRow =
+ SerializationUtils.deserializeBinaryRow(partition.toArray)
+
+ private def serializeSplit(split: DataSplit): Array[Byte] =
SplitSerializer.serialize(split)
+
+ private def deserializeSplit(split: Array[Byte]): DataSplit =
+ SplitSerializer.deserialize(split).asInstanceOf[DataSplit]
+
+}
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 e1301703da..ef49aab883 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
@@ -22,7 +22,6 @@ import org.apache.paimon.{CoreOptions, Snapshot}
import org.apache.paimon.CoreOptions.{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
@@ -30,14 +29,14 @@ import org.apache.paimon.fs.Path
import org.apache.paimon.index.{BucketAssigner, SimpleHashBucketAssigner}
import org.apache.paimon.io.{CompactIncrement, DataIncrement}
import org.apache.paimon.manifest.FileKind
-import org.apache.paimon.spark.{SparkRow, SparkTypeUtils}
+import org.apache.paimon.spark.{SparkPostponeStagedCommitter, SparkRow,
SparkTypeUtils}
import org.apache.paimon.spark.catalog.functions.BucketFunction
import org.apache.paimon.spark.schema.SparkSystemColumns.{BUCKET_COL,
ROW_KIND_COL}
import org.apache.paimon.spark.sort.TableSorter
import org.apache.paimon.spark.util.OptionUtils.paimonExtensionEnabled
import org.apache.paimon.spark.util.SparkRowUtils
import org.apache.paimon.spark.write.{PaimonDataWrite, WriteHelper,
WriteTaskResult}
-import org.apache.paimon.table.{FileStoreTable, PostponeUtils, SpecialFields}
+import org.apache.paimon.table.{FileStoreTable, SpecialFields}
import org.apache.paimon.table.BucketMode._
import org.apache.paimon.table.sink._
import org.apache.paimon.types.{RowKind, RowType}
@@ -49,10 +48,10 @@ import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import java.io.IOException
+import java.util.{Map => JMap}
import java.util.Collections.singletonMap
import scala.collection.JavaConverters._
-import scala.collection.mutable
case class PaimonSparkWriter(
table: FileStoreTable,
@@ -60,8 +59,6 @@ case class PaimonSparkWriter(
batchId: Option[Long] = None)
extends WriteHelper {
- import PaimonSparkWriter._
-
private lazy val tableSchema = table.schema
private lazy val bucketMode = table.bucketMode
@@ -71,6 +68,9 @@ case class PaimonSparkWriter(
@transient private lazy val serializer = new CommitMessageSerializer
+ @transient private var stagedSparkSession: SparkSession = _
+ private var overwritePartitionSpec: Option[Map[String, String]] = None
+
private val writeType = {
if (writeRowTracking) {
// The historical data and new data are processed separately.
@@ -87,12 +87,21 @@ case class PaimonSparkWriter(
val postponeBatchWriteFixedBucket: Boolean =
table.bucketMode() == POSTPONE_MODE &&
coreOptions.postponeBatchWriteFixedBucket()
- val writeBuilder: BatchWriteBuilder = {
- if (postponeBatchWriteFixedBucket) {
- table.newPostponeFixedBucketWriteBuilder()
- } else {
- table.newBatchWriteBuilder()
+ private val postponeBaseSnapshotId =
+ if (postponeBatchWriteFixedBucket)
+ Option(table.snapshotManager().latestSnapshot()).map(_.id())
+ else None
+
+ val writeBuilder: BatchWriteBuilder = table.newBatchWriteBuilder()
+
+ def withOverwrite(): PaimonSparkWriter =
withOverwrite(java.util.Collections.emptyMap())
+
+ def withOverwrite(partition: JMap[String, String]): PaimonSparkWriter = {
+ overwritePartitionSpec = Some(partition.asScala.toMap)
+ if (!postponeBatchWriteFixedBucket) {
+ writeBuilder.withOverwrite(partition)
}
+ this
}
def writeOnly(): PaimonSparkWriter = {
@@ -108,12 +117,6 @@ case class PaimonSparkWriter(
}
def write(data: DataFrame): Seq[CommitMessage] = {
- write(data, overwriteExistingData = false)
- }
-
- private[commands] def write(
- data: DataFrame,
- overwriteExistingData: Boolean): Seq[CommitMessage] = {
val sparkSession = data.sparkSession
val uriReaderFactory = uriReaderFactoryForBlobDescriptor
import sparkSession.implicits._
@@ -126,29 +129,20 @@ case class PaimonSparkWriter(
.withColumn(BUCKET_COL, lit(-1))
case _ => data.withColumn(BUCKET_COL, lit(-1))
}
- val postponeBucketAssignment =
- if (postponeBatchWriteFixedBucket) {
- Some(preparePostponeBucketAssignment(withInitBucketCol,
overwriteExistingData))
- } else {
- None
- }
-
val rowKindColIdx = SparkRowUtils.getFieldIndex(withInitBucketCol.schema,
ROW_KIND_COL)
val bucketColIdx = SparkRowUtils.getFieldIndex(withInitBucketCol.schema,
BUCKET_COL)
val encoderGroupWithBucketCol = EncoderSerDeGroup(withInitBucketCol.schema)
- val postponePartitionBucketComputer =
- postponeBucketAssignment.map(_.partitionBucketComputer)
-
- def newWrite() = PaimonDataWrite(
- writeBuilder,
- writeType,
- rowKindColIdx,
- writeRowTracking,
- fullCompactionDeltaCommits,
- batchId,
- uriReaderFactory,
- postponePartitionBucketComputer
- )
+ def newWrite() =
+ PaimonDataWrite(
+ writeBuilder,
+ writeType,
+ rowKindColIdx,
+ writeRowTracking,
+ fullCompactionDeltaCommits,
+ batchId,
+ uriReaderFactory,
+ None
+ )
def sparkParallelism = {
val defaultParallelism = sparkSession.sparkContext.defaultParallelism
@@ -210,7 +204,7 @@ case class PaimonSparkWriter(
}
}
- val written = bucketMode match {
+ val written: Dataset[_ <: WriteTaskResult] = bucketMode match {
case KEY_DYNAMIC =>
// Topology: input -> bootstrap -> shuffle by key hash ->
bucket-assigner -> shuffle by partition & bucket
val rowType =
SparkTypeUtils.toPaimonType(withInitBucketCol.schema).asInstanceOf[RowType]
@@ -297,16 +291,6 @@ case class PaimonSparkWriter(
)
}
- case POSTPONE_MODE if coreOptions.postponeBatchWriteFixedBucket() =>
- // Topology: input -> bucket-assigner -> shuffle by partition & bucket
- writeWithBucketProcessor(
- withInitBucketCol,
- PostponeFixBucketProcessor(
- table,
- bucketColIdx,
- encoderGroupWithBucketCol,
- postponePartitionBucketComputer.get))
-
case BUCKET_UNAWARE | POSTPONE_MODE =>
var input = data
if (tableSchema.partitionKeys().size() > 0) {
@@ -356,13 +340,11 @@ case class PaimonSparkWriter(
throw new UnsupportedOperationException(s"Spark doesn't support
$bucketMode mode.")
}
- try {
- WriteTaskResult.merge(written.collect())
- } finally {
- if (postponeBucketAssignment.exists(_.dataPersisted)) {
- withInitBucketCol.unpersist()
- }
+ val taskResults = written.collect().toSeq
+ if (postponeBatchWriteFixedBucket) {
+ stagedSparkSession = sparkSession
}
+ WriteTaskResult.merge(taskResults)
}
/**
@@ -437,6 +419,19 @@ case class PaimonSparkWriter(
}
def commit(commitMessages: Seq[CommitMessage], operation:
Snapshot.Operation): Unit = {
+ if (postponeBatchWriteFixedBucket) {
+ if (stagedSparkSession == null) {
+ throw new IllegalStateException("Postpone staged write has no
SparkSession.")
+ }
+ val finalOperation =
Option(operation).getOrElse(Snapshot.Operation.WRITE)
+ val finalMessages = new SparkPostponeStagedCommitter(
+ table,
+ stagedSparkSession,
+ postponeBaseSnapshotId,
+ overwritePartitionSpec).commit(commitMessages, finalOperation)
+ postCommit(finalMessages)
+ return
+ }
val tableCommit = writeBuilder.newCommit()
if (operation != null) {
tableCommit.withOperation(operation)
@@ -540,136 +535,6 @@ case class PaimonSparkWriter(
.toSeq
}
- private def preparePostponeBucketAssignment(
- df: DataFrame,
- overwriteExistingData: Boolean): PostponeBucketAssignment = {
- val knownNumBuckets = PostponeUtils.getKnownNumBuckets(table)
- val maxNumBuckets =
coreOptions.postponeBatchWriteFixedBucketMaxParallelism()
- val unpartitionedTableHasKnownNumBuckets =
- tableSchema.partitionKeys().isEmpty &&
- knownNumBuckets.containsKey(BinaryRow.EMPTY_ROW)
- val inferBucketNumFromData =
- maxNumBuckets != 1 && !unpartitionedTableHasKnownNumBuckets
- if (inferBucketNumFromData) {
- df.persist()
- }
-
- try {
- val defaultNumBuckets = Math.min(df.rdd.getNumPartitions, maxNumBuckets)
- val inferredNumBuckets: Map[BinaryRow, Int] =
- if (inferBucketNumFromData) {
- val targetRowNum = coreOptions.postponeTargetRowNumPerBucket()
- val postponeRowCounts =
- if (overwriteExistingData) {
- java.util.Collections.emptyMap[BinaryRow, java.lang.Long]()
- } else {
- PostponeUtils.getPostponeRowCounts(table)
- }
- val dataStats =
- collectDataStatsByPartition(df, collectSize =
!targetRowNum.isPresent)
- dataStats.map {
- case (partition, stats) =>
- val postponeRowCount = postponeRowCounts.getOrDefault(partition,
0L)
- val numBuckets =
- if (targetRowNum.isPresent) {
- computeBucketNumByRowCount(
- Math.addExact(stats.rowCount, postponeRowCount),
- targetRowNum.get(),
- maxNumBuckets)
- } else {
- computeBucketNumBySize(
- stats,
- postponeRowCount,
- coreOptions.postponeTargetSizePerBucket(),
- maxNumBuckets)
- }
- partition -> numBuckets
- }
- } else {
- Map.empty
- }
- val partitionBucketComputer = (partition: BinaryRow) =>
- knownNumBuckets.getOrDefault(
- partition,
- Integer.valueOf(inferredNumBuckets.getOrElse(partition,
defaultNumBuckets)))
- PostponeBucketAssignment(partitionBucketComputer, inferBucketNumFromData)
- } catch {
- case e: Throwable =>
- if (inferBucketNumFromData) {
- df.unpersist()
- }
- throw e
- }
- }
-
- private def collectDataStatsByPartition(
- df: DataFrame,
- collectSize: Boolean): Map[BinaryRow, PartitionDataStats] = {
- val schema = tableSchema
- val rowType = writeType
- val toPaimonRow = SparkRowUtils.toPaimonRow(
- rowType,
- SparkRowUtils.getFieldIndex(df.schema, ROW_KIND_COL),
- uriReaderFactoryForBlobDescriptor)
- df.rdd
- .mapPartitions {
- rows =>
- val partitionKeyExtractor = new RowPartitionKeyExtractor(schema)
- val rowSerializer = InternalSerializers.create(rowType)
- val stats = mutable.HashMap.empty[SerializedPartition,
PartitionDataStats]
- rows.foreach {
- row =>
- val paimonRow = toPaimonRow(row)
- val partition = SerializedPartition(
-
SerializationUtils.serializeBinaryRow(partitionKeyExtractor.partition(paimonRow)))
- val rowSize =
- if (collectSize)
rowSerializer.toBinaryRow(paimonRow).getSizeInBytes.toLong else 0L
- val previous = stats.getOrElse(partition, PartitionDataStats(0L,
0L))
- stats.put(
- partition,
- PartitionDataStats(
- Math.addExact(previous.rowCount, 1L),
- Math.addExact(previous.serializedSize, rowSize)))
- }
- stats.iterator
- }
- .reduceByKey {
- (left, right) =>
- PartitionDataStats(
- Math.addExact(left.rowCount, right.rowCount),
- Math.addExact(left.serializedSize, right.serializedSize))
- }
- .collect()
- .map {
- case (partition, stats) =>
- SerializationUtils.deserializeBinaryRow(partition.bytes) -> stats
- }
- .toMap
- }
-
- private def computeBucketNumBySize(
- dataStats: PartitionDataStats,
- postponeRowCount: Long,
- targetSizePerBucket: Long,
- maxNumBuckets: Int): Int = {
- if (targetSizePerBucket <= 0) {
- throw new IllegalArgumentException(
- "Option 'postpone.target-size-per-bucket' must be greater than 0.")
- }
-
- // Previous postpone files do not record their uncompressed serialized
size. Estimate it with
- // the average serialized size of incoming rows from the same partition.
- val estimatedTotalSizeNumerator =
- BigInt(dataStats.serializedSize) *
- (BigInt(dataStats.rowCount) + BigInt(postponeRowCount))
- val rowCount = BigInt(dataStats.rowCount)
- val estimatedTotalSize = (estimatedTotalSizeNumerator + rowCount - 1) /
rowCount
- val bucketNum =
- if (estimatedTotalSize == 0) BigInt(1)
- else (estimatedTotalSize - 1) / BigInt(targetSizePerBucket) + 1
- bucketNum.min(BigInt(maxNumBuckets)).toInt
- }
-
private def deserializeCommitMessage(
serializer: CommitMessageSerializer,
bytes: Array[Byte]): CommitMessage = {
@@ -689,38 +554,6 @@ case class PaimonSparkWriter(
}
object PaimonSparkWriter {
-
- private[spark] def computeBucketNumByRowCount(
- rowCount: Long,
- targetRowNumPerBucket: Long,
- maxNumBuckets: Int): Int = {
- if (targetRowNumPerBucket <= 0) {
- throw new IllegalArgumentException(
- "Option 'postpone.target-row-num-per-bucket' must be greater than 0.")
- }
-
- val bucketNum =
- if (rowCount <= 0) 1L else (rowCount - 1) / targetRowNumPerBucket + 1
- Math.min(bucketNum, maxNumBuckets.toLong).toInt
- }
-
- private case class PostponeBucketAssignment(
- partitionBucketComputer: BinaryRow => Integer,
- dataPersisted: Boolean)
-
- private case class PartitionDataStats(rowCount: Long, serializedSize: Long)
-
- private case class SerializedPartition(bytes: Array[Byte]) {
- override def equals(other: Any): Boolean = {
- other match {
- case that: SerializedPartition => java.util.Arrays.equals(bytes,
that.bytes)
- case _ => false
- }
- }
-
- override def hashCode(): Int = java.util.Arrays.hashCode(bytes)
- }
-
def apply(table: FileStoreTable): PaimonSparkWriter = {
new PaimonSparkWriter(table)
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
index 346aa0fe3a..937a47c526 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
@@ -60,7 +60,7 @@ case class WriteIntoPaimonTable(
val writer = PaimonSparkWriter(table, batchId = batchId)
if (overwritePartition != null) {
- writer.writeBuilder.withOverwrite(overwritePartition.asJava)
+ writer.withOverwrite(overwritePartition.asJava)
}
val operation = Option(options.get(PaimonWriteOptions.OPERATION_OPTION))
.map(Snapshot.Operation.valueOf)
@@ -71,10 +71,7 @@ case class WriteIntoPaimonTable(
Snapshot.Operation.WRITE
}
}
- val commitMessages =
- writer.write(
- replacedData,
- overwriteExistingData = overwritePartition != null ||
dynamicPartitionOverwriteMode)
+ val commitMessages = writer.write(replacedData)
writer.commit(commitMessages, operation)
Seq.empty
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 a3141f97b7..25d4f8a91d 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
@@ -213,6 +213,7 @@ case class SparkPostponeCompactProcedure(
realTable.store().newScan(),
realTable.store().newIndexFileHandler(),
snapshotId))
+ dataWrite.write.getWrite().withIgnoreNumBucketCheck(true)
var commitInvoked = false
try {
val pendingBuckets = mutable.LinkedHashMap
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RescaleProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RescaleProcedureTest.scala
index b4a6525116..7d792ca27a 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RescaleProcedureTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RescaleProcedureTest.scala
@@ -84,6 +84,57 @@ class RescaleProcedureTest extends PaimonSparkTestBase {
}
}
+ test("Paimon Procedure: rescale reads deletion-vector level-0 files") {
+ withTable("T") {
+ spark.sql(s"""
+ |CREATE TABLE T (id INT, value STRING)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'id',
+ | 'bucket' = '1',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'deletion-vectors.merge-on-read' = 'false',
+ | 'deletion-vectors.bitmap64' = 'true'
+ |)
+ |""".stripMargin)
+
+ spark.sql("INSERT INTO T VALUES (1, 'one'), (2, 'two'), (3, 'three')")
+ spark.sql("INSERT INTO T VALUES (2, 'updated-two')")
+ val tableWithDv = loadTable("T")
+ val splitsWithDv =
tableWithDv.newSnapshotReader().read().dataSplits().asScala
+ assert(
+ splitsWithDv.exists(
+ _.deletionFiles().orElse(Collections.emptyList()).asScala.exists(_
!= null)))
+
+ // Ordinary non-MOR scans skip level 0. Keep a real level-0 file to
verify that rescale's
+ // explicit split scan is not subject to the ordinary batch-scan pruning.
+ spark.sql("ALTER TABLE T SET TBLPROPERTIES ('write-only' = 'true')")
+ spark.sql("INSERT INTO T VALUES (4, 'level-zero')")
+ assert(
+ loadTable("T")
+ .newSnapshotReader()
+ .read()
+ .dataSplits()
+ .asScala
+ .flatMap(_.dataFiles().asScala)
+ .exists(_.level() == 0))
+ checkAnswer(spark.sql("SELECT * FROM T WHERE id = 4"), Seq.empty)
+
+ spark.sql("ALTER TABLE T SET TBLPROPERTIES ('bucket' = '2', 'write-only'
= 'false')")
+ checkAnswer(spark.sql("CALL sys.rescale(table => 'T', bucket_num =>
2)"), Row(true) :: Nil)
+
+ val rescaledTable = loadTable("T")
+ val rescaledSplits =
rescaledTable.newSnapshotReader().read().dataSplits().asScala
+ assert(rescaledSplits.flatMap(_.dataFiles().asScala).forall(_.level() >
0))
+ assert(
+ rescaledSplits
+ .flatMap(_.deletionFiles().orElse(Collections.emptyList()).asScala)
+ .forall(_ == null))
+ checkAnswer(
+ spark.sql("SELECT * FROM T ORDER BY id"),
+ Seq(Row(1, "one"), Row(2, "updated-two"), Row(3, "three"), Row(4,
"level-zero")))
+ }
+ }
+
test("Paimon Procedure: rescale partitioned tables") {
withTable("T") {
spark.sql(s"""
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 b25c03aec5..a96e0c21e9 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
@@ -18,15 +18,17 @@
package org.apache.paimon.spark.sql
+import org.apache.paimon.Snapshot.CommitKind
import org.apache.paimon.catalog.{Catalog, CatalogLoader, DelegateCatalog,
Identifier}
+import org.apache.paimon.data.BinaryRow
import
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX
import org.apache.paimon.fs.Path
-import org.apache.paimon.spark.{PaimonScan, PaimonSparkTestBase,
PostponeMergeInputScan}
+import org.apache.paimon.spark.{PaimonScan, PaimonSparkTestBase,
PostponeMergeInputScan, SparkTable}
import org.apache.paimon.spark.PaimonMetrics._
-import org.apache.paimon.spark.commands.PaimonSparkWriter
import org.apache.paimon.spark.execution.PostponeMergeOnReadExec
import org.apache.paimon.spark.procedure.SparkPostponeCompactProcedure
-import org.apache.paimon.table.{BucketMode, CatalogEnvironment,
FileStoreTableFactory}
+import org.apache.paimon.table.{BucketMode, CatalogEnvironment,
FileStoreTableFactory, PostponeUtils}
+import org.apache.paimon.table.source.ScanMode
import org.apache.spark.TaskContext
import org.apache.spark.sql.Row
@@ -36,12 +38,474 @@ import scala.collection.JavaConverters._
class PostponeBucketTableTest extends PaimonSparkTestBase {
- test("Postpone bucket table: cap inferred row-count buckets before Int
conversion") {
- assert(
- PaimonSparkWriter.computeBucketNumByRowCount(
- Integer.MAX_VALUE.toLong + 1L,
- targetRowNumPerBucket = 1L,
- maxNumBuckets = 2048) == 2048)
+ test("Postpone bucket table: staged fixed write respects non-ignored empty
commit") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'true',
+ | 'postpone.target-row-num-per-bucket' = '10',
+ | 'snapshot.ignore-empty-commit' = 'false'
+ |)
+ |""".stripMargin)
+
+ sql("INSERT INTO t SELECT CAST(id AS INT), CAST(id AS STRING) FROM
range(0)")
+
+ val latestSnapshot = loadTable("t").latestSnapshot()
+ assert(latestSnapshot.isPresent)
+ assert(latestSnapshot.get().id() == 1L)
+ assert(latestSnapshot.get().deltaRecordCount() == 0L)
+ }
+ }
+
+ test("Postpone bucket table: staged fixed write uses V1 and supports dynamic
partitions") {
+ withTable("t") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> "true") {
+ 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' = '10',
+ | 'postpone.batch-write-fixed-bucket.max-parallelism' = '32'
+ |)
+ |""".stripMargin)
+ assert(!SparkTable(loadTable("t")).useV2Write)
+
+ sql("""
+ |INSERT INTO t SELECT
+ |CAST(id AS INT) AS k,
+ |CAST(id AS STRING) AS v,
+ |CASE WHEN id < 5 THEN 0 WHEN id < 26 THEN 1 ELSE 2 END AS pt
+ |FROM range(0, 68)
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT pt, count(*) FROM t GROUP BY pt ORDER BY pt"),
+ Seq(Row(0, 5L), Row(1, 21L), Row(2, 42L)))
+ val knownBuckets =
PostponeUtils.getKnownNumBuckets(loadTable("t")).asScala
+ assert(knownBuckets(BinaryRow.singleColumn(0)) == 1)
+ assert(knownBuckets(BinaryRow.singleColumn(1)) == 4)
+ assert(knownBuckets(BinaryRow.singleColumn(2)) == 8)
+ checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket = -2"),
Seq(Row(0L)))
+ assert(loadTable("t").latestSnapshot().get().id() == 1L)
+ }
+ }
+ }
+
+ test("Postpone bucket table: staged rescale supports per-partition layouts")
{
+ 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',
+ | 'postpone.batch-write-fixed-bucket.max-parallelism' = '64'
+ |)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO t
+ |SELECT 0 AS k, 'p0-initial' AS v, 0 AS pt
+ |UNION ALL
+ |SELECT CAST(id AS INT), CAST(id AS STRING), 1 AS pt FROM range(0,
8)
+ |UNION ALL
+ |SELECT 0 AS k, 'p3-initial' AS v, 3 AS pt
+ |""".stripMargin)
+
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" ->
"false") {
+ sql("""INSERT INTO t VALUES
+ |(0, 'historical-update', 0),
+ |(9999, 'historical-only', 0),
+ |(8888, 'untouched-postpone', 2)
+ |""".stripMargin)
+ }
+ checkAnswer(
+ sql("SELECT count(*) FROM `t$buckets` WHERE partition = '{0}' AND
bucket = -2"),
+ Seq(Row(1L)))
+ checkAnswer(
+ sql("SELECT count(*) FROM `t$buckets` WHERE partition = '{2}' AND
bucket = -2"),
+ Seq(Row(1L)))
+
+ sql("""
+ |INSERT INTO t
+ |SELECT CAST(id AS INT), CONCAT('current-', CAST(id AS STRING)), 0
AS pt
+ |FROM range(0, 130)
+ |UNION ALL
+ |SELECT 100 AS k, 'p1-new' AS v, 1 AS pt
+ |UNION ALL
+ |SELECT CAST(id AS INT), CONCAT('p3-', CAST(id AS STRING)), 3 AS pt
+ |FROM range(0, 33)
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT pt, count(*) FROM t GROUP BY pt ORDER BY pt"),
+ Seq(Row(0, 130L), Row(1, 9L), Row(3, 33L)))
+ checkAnswer(
+ sql("SELECT k, v FROM t WHERE pt = 0 AND k IN (0, 9999) ORDER BY k"),
+ Seq(Row(0, "current-0")))
+ val knownBuckets =
PostponeUtils.getKnownNumBuckets(loadTable("t")).asScala
+ assert(knownBuckets(BinaryRow.singleColumn(0)) == 64)
+ assert(knownBuckets(BinaryRow.singleColumn(1)) == 8)
+ assert(knownBuckets(BinaryRow.singleColumn(3)) == 64)
+ checkAnswer(
+ sql("SELECT count(*) FROM `t$buckets` WHERE partition = '{0}' AND
bucket = -2"),
+ Seq(Row(1L)))
+ checkAnswer(
+ sql("SELECT count(*) FROM `t$buckets` WHERE partition = '{2}' AND
bucket = -2"),
+ Seq(Row(1L)))
+ assert(loadTable("t").latestSnapshot().get().id() == 4L)
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(
+ sql("SELECT k, v, pt FROM t WHERE k IN (8888, 9999) ORDER BY k"),
+ Seq(Row(8888, "untouched-postpone", 2), Row(9999, "historical-only",
0)))
+ }
+ }
+ }
+
+ test("Postpone bucket table: load-factor rescale") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k',
+ | 'bucket' = '-2',
+ | 'changelog-producer' = 'input',
+ | 'postpone.batch-write-fixed-bucket' = 'true',
+ | 'postpone.target-row-num-per-bucket' = '1',
+ | 'postpone.batch-write-fixed-bucket.max-parallelism' = '16',
+ | 'postpone.batch-write-fixed-bucket.rescale-load-factor' = '32'
+ |)
+ |""".stripMargin)
+
+ sql("INSERT INTO t VALUES (0, 'initial')")
+
assert(PostponeUtils.getKnownNumBuckets(loadTable("t")).get(BinaryRow.EMPTY_ROW)
== 1)
+
+ sql("""
+ |INSERT INTO t
+ |SELECT CAST(id AS INT), CAST(id AS STRING) FROM range(1, 33)
+ |""".stripMargin)
+
assert(PostponeUtils.getKnownNumBuckets(loadTable("t")).get(BinaryRow.EMPTY_ROW)
== 1)
+
+ sql("""
+ |INSERT INTO t
+ |SELECT CAST(id AS INT), CAST(id AS STRING) FROM range(33, 66)
+ |""".stripMargin)
+
+ checkAnswer(sql("SELECT count(*), sum(k) FROM t"), Seq(Row(66L, 2145L)))
+ val resultTable = loadTable("t")
+
assert(PostponeUtils.getKnownNumBuckets(resultTable).get(BinaryRow.EMPTY_ROW)
== 16)
+ checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket = -2"),
Seq(Row(0L)))
+ assert(resultTable.latestSnapshot().get().id() == 4L)
+ assert(resultTable.snapshotManager().snapshot(3L).commitKind() ==
CommitKind.OVERWRITE)
+
assert(resultTable.snapshotManager().snapshot(3L).changelogManifestList() ==
null)
+ assert(resultTable.snapshotManager().snapshot(4L).commitKind() ==
CommitKind.APPEND)
+ assert(
+ !resultTable
+ .newSnapshotReader()
+ .withMode(ScanMode.CHANGELOG)
+ .read()
+ .dataSplits()
+ .isEmpty)
+ }
+ }
+
+ test("Postpone bucket table: staged fixed write does not compact") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k',
+ | 'bucket' = '1',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'deletion-vectors.merge-on-read' = 'true',
+ | 'deletion-vectors.bitmap64' = 'false'
+ |)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO t
+ |SELECT CAST(id AS INT), CONCAT('base-', CAST(id AS STRING)) FROM
range(0, 4)
+ |""".stripMargin)
+ sql("INSERT INTO t VALUES (3, 'updated-3')")
+ assert(deletionVectorCardinality("t") == 1L)
+ checkAnswer(
+ sql("SELECT * FROM t ORDER BY k"),
+ Seq(Row(0, "base-0"), Row(1, "base-1"), Row(2, "base-2"), Row(3,
"updated-3")))
+
+ sql("""
+ |ALTER TABLE t SET TBLPROPERTIES (
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'true',
+ | 'postpone.target-row-num-per-bucket' = '4',
+ | 'postpone.batch-write-fixed-bucket.max-parallelism' = '8',
+ | 'postpone.batch-write-fixed-bucket.rescale-load-factor' = '2'
+ |)
+ |""".stripMargin)
+
assert(PostponeUtils.getKnownNumBuckets(loadTable("t")).get(BinaryRow.EMPTY_ROW)
== 1)
+
+ // The fixed writer restores the old bucket but remains write-only.
Existing deletion
+ // vectors and new level-0 files are left for background compaction.
+ sql("""
+ |INSERT INTO t
+ |SELECT 2 AS k, 'updated-2' AS v
+ |UNION ALL
+ |SELECT CAST(id AS INT), CONCAT('new-', CAST(id AS STRING)) FROM
range(4, 8)
+ |""".stripMargin)
+
+
assert(PostponeUtils.getKnownNumBuckets(loadTable("t")).get(BinaryRow.EMPTY_ROW)
== 1)
+ assert(deletionVectorCardinality("t") == 1L)
+ assert(
+ loadTable("t")
+ .newSnapshotReader()
+ .onlyReadRealBuckets()
+ .read()
+ .dataSplits()
+ .asScala
+ .flatMap(_.dataFiles().asScala)
+ .exists(_.level() == 0))
+ checkAnswer(
+ sql("SELECT * FROM t ORDER BY k"),
+ Seq(
+ Row(0, "base-0"),
+ Row(1, "base-1"),
+ Row(2, "updated-2"),
+ Row(3, "updated-3"),
+ Row(4, "new-4"),
+ Row(5, "new-5"),
+ Row(6, "new-6"),
+ Row(7, "new-7"))
+ )
+ checkAnswer(sql("SELECT count(*) FROM t"), Seq(Row(8L)))
+ checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket = -2"),
Seq(Row(0L)))
+ }
+ }
+
+ test("Postpone bucket table: staged rescale reads level 0 and applies
deletion vectors") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING,
+ | pt INT
+ |) PARTITIONED BY (pt)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'k, pt',
+ | 'bucket' = '1',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'deletion-vectors.merge-on-read' = 'false',
+ | 'deletion-vectors.bitmap64' = 'true'
+ |)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO t
+ |SELECT CAST(id AS INT), CONCAT('base-', CAST(id AS STRING)), 0 AS
pt
+ |FROM range(0, 4)
+ |UNION ALL
+ |SELECT CAST(id AS INT), CONCAT('base-', CAST(id AS STRING)), 1 AS
pt
+ |FROM range(0, 4)
+ |""".stripMargin)
+ sql("INSERT INTO t VALUES (2, 'updated-2', 0), (2, 'updated-2', 1)")
+ assert(deletionVectorCardinality("t") == 2L)
+ assert(deletionVectorCardinality("t", BinaryRow.singleColumn(0)) == 1L)
+ assert(deletionVectorCardinality("t", BinaryRow.singleColumn(1)) == 1L)
+ checkAnswer(
+ sql("SELECT k, v, pt FROM t ORDER BY pt, k"),
+ Seq(
+ Row(0, "base-0", 0),
+ Row(1, "base-1", 0),
+ Row(2, "updated-2", 0),
+ Row(3, "base-3", 0),
+ Row(0, "base-0", 1),
+ Row(1, "base-1", 1),
+ Row(2, "updated-2", 1),
+ Row(3, "base-3", 1)
+ )
+ )
+
+ // Ordinary batch scans skip DV level-0 files when merge-on-read is
false. Keep one such
+ // file in partition 0 to verify that rescale still reads the complete
logical snapshot.
+ sql("ALTER TABLE t SET TBLPROPERTIES ('write-only' = 'true')")
+ sql("INSERT INTO t VALUES (50, 'level-0', 0)")
+ val partition0 = BinaryRow.singleColumn(0)
+ assert(
+ loadTable("t")
+ .newSnapshotReader()
+ .onlyReadRealBuckets()
+ .read()
+ .dataSplits()
+ .asScala
+ .filter(_.partition() == partition0)
+ .flatMap(_.dataFiles().asScala)
+ .exists(_.level() == 0))
+ checkAnswer(sql("SELECT * FROM t WHERE k = 50 AND pt = 0"), Seq.empty)
+
+ sql("""
+ |ALTER TABLE t SET TBLPROPERTIES (
+ | 'bucket' = '-2',
+ | 'write-only' = 'false',
+ | 'postpone.batch-write-fixed-bucket' = 'true',
+ | 'postpone.target-row-num-per-bucket' = '4',
+ | 'postpone.batch-write-fixed-bucket.max-parallelism' = '8',
+ | 'postpone.batch-write-fixed-bucket.rescale-load-factor' = '2'
+ |)
+ |""".stripMargin)
+ val initialBuckets = PostponeUtils.getKnownNumBuckets(loadTable("t"))
+ assert(initialBuckets.get(partition0) == 1)
+ assert(initialBuckets.get(BinaryRow.singleColumn(1)) == 1)
+ assert(!loadTable("t").coreOptions().writeOnly())
+ assert(loadTable("t").coreOptions().needLookup())
+
+ // Only partition 0 exceeds the load factor. Its old file and deletion
vectors are
+ // materialized into the new layout, while partition 1 and its deletion
vectors stay intact.
+ sql("""
+ |INSERT INTO t
+ |SELECT 1 AS k, 'new-1' AS v, 0 AS pt
+ |UNION ALL
+ |SELECT CAST(id AS INT), CONCAT('new-', CAST(id AS STRING)), 0 AS
pt
+ |FROM range(100, 108)
+ |""".stripMargin)
+
+ val resultBuckets = PostponeUtils.getKnownNumBuckets(loadTable("t"))
+ assert(resultBuckets.get(partition0) == 4)
+ assert(resultBuckets.get(BinaryRow.singleColumn(1)) == 1)
+ assert(deletionVectorCardinality("t") == 1L)
+ assert(deletionVectorCardinality("t", partition0) == 0L)
+ assert(deletionVectorCardinality("t", BinaryRow.singleColumn(1)) == 1L)
+ val resultFiles = loadTable("t")
+ .newSnapshotReader()
+ .onlyReadRealBuckets()
+ .read()
+ .dataSplits()
+ .asScala
+ .filter(_.partition() == partition0)
+ .flatMap(_.dataFiles().asScala)
+ assert(
+ resultFiles.exists(_.level() == 0),
+ resultFiles.map(file =>
s"${file.fileName()}:L${file.level()}").mkString(", "))
+ val expected =
+ Seq(
+ Row(0, "base-0", 0),
+ Row(1, "new-1", 0),
+ Row(2, "updated-2", 0),
+ Row(3, "base-3", 0),
+ Row(50, "level-0", 0)) ++
+ (100 until 108).map(id => Row(id, s"new-$id", 0)) ++
+ Seq(Row(0, "base-0", 1), Row(1, "base-1", 1), Row(2, "updated-2",
1), Row(3, "base-3", 1))
+ withSparkSQLConf("spark.paimon.deletion-vectors.merge-on-read" ->
"true") {
+ checkAnswer(sql("SELECT k, v, pt FROM t ORDER BY pt, k"), expected)
+ checkAnswer(
+ sql("SELECT pt, count(*) FROM t GROUP BY pt ORDER BY pt"),
+ Seq(Row(0, 13L), Row(1, 4L)))
+ }
+
+ // Background compaction materializes the level-0 files and deletion
vectors.
+ sql("CALL sys.compact(table => 't')")
+ val compactedFiles = loadTable("t")
+ .newSnapshotReader()
+ .onlyReadRealBuckets()
+ .read()
+ .dataSplits()
+ .asScala
+ .flatMap(_.dataFiles().asScala)
+ assert(
+ compactedFiles.forall(_.level() > 0),
+ compactedFiles.map(file =>
s"${file.fileName()}:L${file.level()}").mkString(", "))
+ checkAnswer(sql("SELECT k, v, pt FROM t ORDER BY pt, k"), expected)
+ checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket = -2"),
Seq(Row(0L)))
+ }
+ }
+
+ test("Postpone bucket table: staged write preserves historical postpone
without rescale") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'true',
+ | 'postpone.target-row-num-per-bucket' = '1',
+ | 'postpone.batch-write-fixed-bucket.max-parallelism' = '16'
+ |)
+ |""".stripMargin)
+
+ sql("INSERT INTO t SELECT CAST(id AS INT), CAST(id AS STRING) FROM
range(0, 8)")
+
assert(PostponeUtils.getKnownNumBuckets(loadTable("t")).get(BinaryRow.EMPTY_ROW)
== 8)
+
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" ->
"false") {
+ sql("INSERT INTO t VALUES (0, 'historical-update'), (100,
'historical-only')")
+ }
+ sql("INSERT INTO t VALUES (0, 'current-update')")
+
+ checkAnswer(sql("SELECT count(*) FROM t"), Seq(Row(8L)))
+ checkAnswer(
+ sql("SELECT * FROM t WHERE k IN (0, 100) ORDER BY k"),
+ Seq(Row(0, "current-update")))
+ checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket = -2"),
Seq(Row(1L)))
+
assert(PostponeUtils.getKnownNumBuckets(loadTable("t")).get(BinaryRow.EMPTY_ROW)
== 8)
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(sql("SELECT * FROM t WHERE k = 100"), Seq(Row(100,
"historical-only")))
+ }
+ }
+ }
+
+ test("Postpone bucket table: real-bucket rescale preserves historical
postpone data") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING,
+ | pt INT
+ |) PARTITIONED BY (pt)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'k, pt',
+ | 'bucket' = '-2',
+ | 'dynamic-partition-overwrite' = 'true',
+ | 'postpone.batch-write-fixed-bucket' = 'true',
+ | 'postpone.target-row-num-per-bucket' = '1',
+ | 'postpone.batch-write-fixed-bucket.max-parallelism' = '16'
+ |)
+ |""".stripMargin)
+
+ sql("INSERT INTO t VALUES (1, 'base', 0)")
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" ->
"false") {
+ sql("DELETE FROM t WHERE k = 1 AND pt = 0")
+ }
+ sql("""
+ |INSERT INTO t
+ |SELECT CAST(id AS INT), CAST(id AS STRING), 0 AS pt FROM range(2,
35)
+ |""".stripMargin)
+
+ checkAnswer(sql("SELECT count(*) FROM t"), Seq(Row(34L)))
+
assert(PostponeUtils.getKnownNumBuckets(loadTable("t")).get(BinaryRow.singleColumn(0))
== 16)
+ checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket = -2"),
Seq(Row(1L)))
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(sql("SELECT count(*) FROM t"), Seq(Row(33L)))
+ checkAnswer(sql("SELECT * FROM t WHERE k = 1"), Seq.empty)
+ }
+ }
}
test("Postpone bucket table: write with different bucket number") {
@@ -99,7 +563,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
|""".stripMargin)
checkAnswer(
sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{5}'
ORDER BY bucket"),
- Seq(Row(0), Row(1), Row(2))
+ Seq(Row(0), Row(1))
)
}
}
@@ -134,7 +598,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
)
checkAnswer(
sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{1}'
ORDER BY bucket"),
- Seq(Row(0), Row(1), Row(2))
+ Seq(Row(0), Row(1), Row(2), Row(3))
)
// Existing partitions keep their bucket number even when the new data
volume changes.
@@ -150,7 +614,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
Seq(Row(0))
)
- // Postpone rows are included when a partition gets real buckets for the
first time.
+ // Historical postpone rows neither participate in inference nor enter
the fixed write.
withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" ->
"false") {
sql("""
|INSERT INTO t SELECT
@@ -169,8 +633,13 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
|""".stripMargin)
checkAnswer(
sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{2}'
ORDER BY bucket"),
- Seq(Row(-2), Row(0), Row(1))
+ Seq(Row(-2), Row(0))
)
+ checkAnswer(sql("SELECT count(*) FROM t WHERE pt = 2"), Seq(Row(100L)))
+
assert(PostponeUtils.getKnownNumBuckets(loadTable("t")).get(BinaryRow.singleColumn(2))
== 1)
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(sql("SELECT count(*) FROM t WHERE pt = 2"), Seq(Row(250L)))
+ }
}
}
@@ -205,43 +674,47 @@ class PostponeBucketTableTest extends PaimonSparkTestBase
{
).foreach {
case (partitionOverwriteMode, overwriteSql) =>
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' = 'false',
- | 'postpone.target-row-num-per-bucket' = '100'
- |)
- |""".stripMargin)
-
- sql("""
- |INSERT INTO t SELECT
- |id AS k,
- |CAST(id AS STRING) AS v,
- |0 AS pt
- |FROM range (0, 1000)
- |""".stripMargin)
-
- withSparkSQLConf(
- "spark.paimon.postpone.batch-write-fixed-bucket" -> "true",
- "spark.sql.sources.partitionOverwriteMode" ->
partitionOverwriteMode) {
- sql(overwriteSql)
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> "true") {
+ 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' = 'false',
+ | 'postpone.target-row-num-per-bucket' = '100'
+ |)
+ |""".stripMargin)
+ assert(SparkTable(loadTable("t")).useV2Write)
+
+ sql("""
+ |INSERT INTO t SELECT
+ |id AS k,
+ |CAST(id AS STRING) AS v,
+ |0 AS pt
+ |FROM range (0, 1000)
+ |""".stripMargin)
+
+ withSparkSQLConf(
+ "spark.paimon.postpone.batch-write-fixed-bucket" -> "true",
+ "spark.sql.sources.partitionOverwriteMode" ->
partitionOverwriteMode) {
+ sql(overwriteSql)
+ }
+
+ checkAnswer(sql("SELECT count(*), sum(k) FROM t"), Seq(Row(100L,
104950L)))
+ checkAnswer(
+ sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition =
'{0}'"),
+ Seq(Row(0)))
+ checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket =
-2"), Seq(Row(0L)))
}
-
- checkAnswer(
- sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition =
'{0}'"),
- Seq(Row(0))
- )
}
}
}
- test("Postpone bucket table: infer bucket number from serialized data size")
{
+ test("Postpone bucket table: infer bucket number from staged file size") {
withTable("t") {
sql("""
|CREATE TABLE t (
@@ -258,43 +731,45 @@ class PostponeBucketTableTest extends PaimonSparkTestBase
{
|""".stripMargin)
sql("""
- |INSERT INTO t SELECT /*+ REPARTITION(20) */
+ |INSERT INTO t SELECT /*+ REPARTITION(1) */
|id AS k,
- |CASE WHEN id < 100 THEN repeat('x', 100) ELSE repeat('x', 1000)
END AS v,
- |CASE WHEN id < 100 THEN 0 ELSE 1 END AS pt
- |FROM range (0, 200)
+ |CASE WHEN id < 10 THEN sha2(CAST(id AS STRING), 256)
+ | ELSE array_join(transform(sequence(0, 63),
+ | x -> sha2(concat(CAST(id AS STRING), '-', CAST(x AS
STRING)), 256)), '')
+ |END AS v,
+ |CASE WHEN id < 10 THEN 0 ELSE 1 END AS pt
+ |FROM range (0, 110)
|""".stripMargin)
- checkAnswer(
- sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{0}'
ORDER BY bucket"),
- Seq(Row(0))
- )
- checkAnswer(
- sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{1}'
ORDER BY bucket"),
- Seq(Row(0), Row(1), Row(2), Row(3))
- )
+ val initialBuckets = PostponeUtils.getKnownNumBuckets(loadTable("t"))
+ assert(initialBuckets.get(BinaryRow.singleColumn(0)) == 1)
+ assert(initialBuckets.get(BinaryRow.singleColumn(1)) > 1)
- // Estimate existing postpone data with the average serialized size of
incoming rows.
+ // Historical postpone data is ignored; only the exactly measured
current batch is inferred.
withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" ->
"false") {
sql("""
|INSERT INTO t SELECT
|id AS k,
- |repeat('x', 1000) AS v,
+ |array_join(transform(sequence(0, 63),
+ | x -> sha2(concat(CAST(id AS STRING), '-', CAST(x AS STRING)),
256)), '') AS v,
|2 AS pt
|FROM range (1000, 1100)
|""".stripMargin)
}
sql("""
- |INSERT INTO t SELECT /*+ REPARTITION(20) */
+ |INSERT INTO t SELECT /*+ REPARTITION(1) */
|id AS k,
- |repeat('x', 1000) AS v,
+ |array_join(transform(sequence(0, 63),
+ | x -> sha2(concat(CAST(id AS STRING), '-', CAST(x AS STRING)),
256)), '') AS v,
|2 AS pt
|FROM range (2000, 2100)
|""".stripMargin)
- checkAnswer(
- sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{2}'
ORDER BY bucket"),
- Seq(Row(-2), Row(0), Row(1), Row(2), Row(3), Row(4), Row(5), Row(6))
- )
+
assert(PostponeUtils.getKnownNumBuckets(loadTable("t")).get(BinaryRow.singleColumn(2))
> 1)
+ checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket = -2"),
Seq(Row(1L)))
+ checkAnswer(sql("SELECT count(*) FROM t WHERE pt = 2"), Seq(Row(100L)))
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(sql("SELECT count(*) FROM t WHERE pt = 2"), Seq(Row(200L)))
+ }
}
}
@@ -357,6 +832,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
| 'deletion-vectors.enabled' = 'true',
| 'deletion-vectors.merge-on-read' = 'true',
| 'postpone.default-bucket-num' = '1',
+ | 'postpone.batch-write-fixed-bucket.max-parallelism' = '1',
| 'source.split.target-size' = '1 B'
|)
|""".stripMargin)
@@ -722,6 +1198,19 @@ class PostponeBucketTableTest extends PaimonSparkTestBase
{
// Core must retain seq after Spark projects v.
checkAnswer(sql("SELECT v FROM t ORDER BY v"), Seq(Row("base-1"),
Row("newer")))
}
+
+ // The fixed write ignores historical postpone rows. They still
participate when merge-on-read
+ // is enabled, where the user sequence field takes precedence.
+ sql("INSERT INTO t VALUES (1, 'current', 15), (2, 'current-older', 15)")
+ checkAnswer(
+ sql("SELECT * FROM t ORDER BY k"),
+ Seq(Row(1, "current", 15), Row(2, "current-older", 15)))
+ checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket = -2"),
Seq(Row(1L)))
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(
+ sql("SELECT * FROM t ORDER BY k"),
+ Seq(Row(1, "current", 15), Row(2, "newer", 20)))
+ }
}
}
@@ -863,6 +1352,19 @@ class PostponeBucketTableTest extends PaimonSparkTestBase
{
checkAnswer(sql("SELECT * FROM partial_t WHERE v2 = 'b'"), Seq(Row(1,
"a", "b")))
checkAnswer(sql("SELECT * FROM aggregation_t WHERE total = 13"),
Seq(Row(1, 13L)))
}
+
+ sql("INSERT INTO partial_t VALUES (1, 'c', CAST(NULL AS STRING))")
+ sql("INSERT INTO aggregation_t VALUES (1, 7L)")
+ checkAnswer(sql("SELECT * FROM partial_t"), Seq(Row(1, "c", null)))
+ checkAnswer(sql("SELECT * FROM aggregation_t"), Seq(Row(1, 17L)))
+ checkAnswer(sql("SELECT count(*) FROM `partial_t$buckets` WHERE bucket =
-2"), Seq(Row(1L)))
+ checkAnswer(
+ sql("SELECT count(*) FROM `aggregation_t$buckets` WHERE bucket = -2"),
+ Seq(Row(1L)))
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(sql("SELECT * FROM partial_t"), Seq(Row(1, "c", "b")))
+ checkAnswer(sql("SELECT * FROM aggregation_t"), Seq(Row(1, 20L)))
+ }
}
}
@@ -1153,17 +1655,27 @@ class PostponeBucketTableTest extends
PaimonSparkTestBase {
}
private def deletionVectorCardinality(tableName: String): Long = {
+ deletionVectorCardinality(tableName, None)
+ }
+
+ private def deletionVectorCardinality(tableName: String, partition:
BinaryRow): Long = {
+ deletionVectorCardinality(tableName, Some(partition))
+ }
+
+ private def deletionVectorCardinality(tableName: String, partition:
Option[BinaryRow]): Long = {
val table = loadTable(tableName)
table
.store()
.newIndexFileHandler()
.scan(table.latestSnapshot().get(), DELETION_VECTORS_INDEX)
.asScala
+ .filter(entry => partition.forall(_ == entry.partition()))
.flatMap(entry => Option(entry.indexFile().dvRanges()).toSeq)
.flatMap(_.values().asScala)
.flatMap(meta => Option(meta.cardinality()).map(_.longValue()))
.sum
}
+
}
object PostponeBucketTableTest {
