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 2281336cc8 [core][flink][spark] Introduce postpone fixed-bucket write
builder (#8992)
2281336cc8 is described below
commit 2281336cc87a7e4dbd050bb22dcc9f669583ac50
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Aug 3 15:35:42 2026 +0800
[core][flink][spark] Introduce postpone fixed-bucket write builder (#8992)
---
.../java/org/apache/paimon/KeyValueFileStore.java | 27 +++++--
.../paimon/operation/AbstractFileStoreWrite.java | 83 ++++++++++++++++++----
.../paimon/operation/FileStoreCommitImpl.java | 16 +++--
.../apache/paimon/operation/FileStoreWrite.java | 10 +++
.../paimon/operation/commit/ConflictDetection.java | 48 +++++++------
.../paimon/privilege/PrivilegedFileStoreTable.java | 7 ++
.../paimon/table/DelegatedFileStoreTable.java | 6 ++
.../org/apache/paimon/table/FileStoreTable.java | 13 ++++
.../org/apache/paimon/table/PostponeUtils.java | 14 ----
.../paimon/table/PrimaryKeyFileStoreTable.java | 13 +++-
.../paimon/table/sink/BatchWriteBuilderImpl.java | 19 -----
...l.java => PostponeFixedBucketWriteBuilder.java} | 68 ++++++++----------
.../apache/paimon/table/sink/TableWriteImpl.java | 22 +++++-
.../paimon/operation/FileStoreCommitTest.java | 64 +++++++++++++++++
.../paimon/table/PrimaryKeySimpleTableTest.java | 55 ++++++++++++++
.../org/apache/paimon/flink/sink/FlinkSink.java | 24 +++++--
.../apache/paimon/flink/sink/FlinkSinkBuilder.java | 5 +-
.../flink/sink/PostponeBatchWriteOperator.java | 29 +-------
.../sink/PostponeFixedBucketChannelComputer.java | 8 ++-
.../paimon/flink/sink/PostponeFixedBucketSink.java | 47 ++++++------
.../apache/paimon/flink/sink/StoreSinkWrite.java | 28 ++++++++
.../paimon/flink/sink/StoreSinkWriteImpl.java | 42 ++++++++++-
.../PostponeFixedBucketChannelComputerTest.java | 30 ++++++++
.../flink/sink/StoreCompactOperatorTest.java | 5 ++
.../paimon/spark/commands/PaimonSparkWriter.scala | 18 ++---
.../paimon/spark/write/PaimonDataWrite.scala | 29 +++-----
26 files changed, 506 insertions(+), 224 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
index b3c06c98c1..8e072a0736 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
@@ -33,6 +33,7 @@ import org.apache.paimon.operation.KeyValueFileStoreScan;
import org.apache.paimon.operation.KeyValueFileStoreWrite;
import org.apache.paimon.operation.MergeFileSplitRead;
import org.apache.paimon.operation.RawFileSplitRead;
+import org.apache.paimon.options.Options;
import org.apache.paimon.postpone.PostponeBucketFileStoreWrite;
import org.apache.paimon.schema.KeyValueFieldsExtractor;
import org.apache.paimon.schema.SchemaManager;
@@ -162,19 +163,31 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
tableName,
writeId);
}
+ return newFixedBucketWrite(commitUser, options);
+ }
+
+ /** Creates a merge-tree writer for fixed-bucket batch writes to a
postpone-bucket table. */
+ public AbstractFileStoreWrite<KeyValue> newPostponeFixedBucketWrite(String
commitUser) {
+ Options writeOptions = new Options(options.toMap());
+ writeOptions.set(CoreOptions.WRITE_ONLY, true);
+ return newFixedBucketWrite(commitUser, new CoreOptions(writeOptions));
+ }
+
+ private AbstractFileStoreWrite<KeyValue> newFixedBucketWrite(
+ String commitUser, CoreOptions writeOptions) {
DynamicBucketIndexMaintainer.Factory indexFactory = null;
if (bucketMode() == BucketMode.HASH_DYNAMIC) {
indexFactory = new
DynamicBucketIndexMaintainer.Factory(newIndexFileHandler());
}
BucketedDvMaintainer.Factory dvMaintainerFactory = null;
- if (options.deletionVectorsEnabled()) {
+ if (writeOptions.deletionVectorsEnabled()) {
dvMaintainerFactory =
BucketedDvMaintainer.factory(newIndexFileHandler());
}
BucketedPrimaryKeyIndexMaintainer.Factory
primaryKeyIndexMaintainerFactory = null;
- if (options.primaryKeyVectorIndexEnabled()
- || options.primaryKeyFullTextIndexEnabled()
- || !options.primaryKeyBTreeIndexColumns().isEmpty()
- || !options.primaryKeyBitmapIndexColumns().isEmpty()) {
+ if (writeOptions.primaryKeyVectorIndexEnabled()
+ || writeOptions.primaryKeyFullTextIndexEnabled()
+ || !writeOptions.primaryKeyBTreeIndexColumns().isEmpty()
+ || !writeOptions.primaryKeyBitmapIndexColumns().isEmpty()) {
primaryKeyIndexMaintainerFactory =
BucketedPrimaryKeyIndexMaintainer.Factory.create(
newIndexFileHandler(), newReaderFactoryBuilder(),
schema);
@@ -188,7 +201,7 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
keyType,
valueType,
keyComparatorSupplier,
- () -> UserDefinedSeqComparator.create(valueType, options),
+ () -> UserDefinedSeqComparator.create(valueType, writeOptions),
logDedupEqualSupplier,
mfFactory,
pathFactory(),
@@ -198,7 +211,7 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
indexFactory,
dvMaintainerFactory,
primaryKeyIndexMaintainerFactory,
- options,
+ writeOptions,
keyValueFieldsExtractor,
tableName);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
index e83f5e81c1..2ee11440dc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
@@ -72,6 +72,7 @@ import static
org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME;
import static org.apache.paimon.io.DataFileMeta.getMaxSequenceNumber;
import static
org.apache.paimon.shade.guava30.com.google.common.base.MoreObjects.firstNonNull;
import static
org.apache.paimon.utils.FileStorePathFactory.getPartitionComputer;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
/**
* Base {@link FileStoreWrite} implementation.
@@ -185,6 +186,22 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
@Override
public void write(BinaryRow partition, int bucket, T data) throws
Exception {
WriterContainer<T> container = getWriterWrapper(partition, bucket);
+ write(container, data);
+ }
+
+ @Override
+ public void write(BinaryRow partition, int bucket, int totalBuckets, T
data) throws Exception {
+ checkArgument(totalBuckets > 0, "Total number of buckets must be
positive.");
+ checkArgument(
+ bucket >= 0 && bucket < totalBuckets,
+ "Bucket %s is out of range [0, %s).",
+ bucket,
+ totalBuckets);
+ WriterContainer<T> container = getWriterWrapper(partition, bucket,
totalBuckets);
+ write(container, data);
+ }
+
+ private void write(WriterContainer<T> container, T data) throws Exception {
container.writer.write(data);
if (container.dynamicBucketMaintainer != null) {
container.dynamicBucketMaintainer.notifyNewRecord((KeyValue) data);
@@ -460,13 +477,28 @@ public abstract class AbstractFileStoreWrite<T>
implements FileStoreWrite<T> {
}
protected WriterContainer<T> getWriterWrapper(BinaryRow partition, int
bucket) {
+ Map<Integer, WriterContainer<T>> buckets =
getWriterContainers(partition);
+ return buckets.computeIfAbsent(
+ bucket, k -> createWriterContainer(partition.copy(), bucket));
+ }
+
+ private WriterContainer<T> getWriterWrapper(BinaryRow partition, int
bucket, int totalBuckets) {
+ Map<Integer, WriterContainer<T>> buckets =
getWriterContainers(partition);
+ if (!buckets.isEmpty()) {
+ checkNumBuckets(
+ partition, totalBuckets,
buckets.values().iterator().next().totalBuckets);
+ }
+ return buckets.computeIfAbsent(
+ bucket, k -> createWriterContainer(partition.copy(), bucket,
totalBuckets));
+ }
+
+ private Map<Integer, WriterContainer<T>> getWriterContainers(BinaryRow
partition) {
Map<Integer, WriterContainer<T>> buckets = writers.get(partition);
if (buckets == null) {
buckets = new HashMap<>();
writers.put(partition.copy(), buckets);
}
- return buckets.computeIfAbsent(
- bucket, k -> createWriterContainer(partition.copy(), bucket));
+ return buckets;
}
public RecordWriter<T> createWriter(BinaryRow partition, int bucket) {
@@ -474,6 +506,16 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
}
public WriterContainer<T> createWriterContainer(BinaryRow partition, int
bucket) {
+ return createWriterContainer(partition, bucket, numBuckets,
!ignoreNumBucketCheck);
+ }
+
+ private WriterContainer<T> createWriterContainer(
+ BinaryRow partition, int bucket, int totalBuckets) {
+ return createWriterContainer(partition, bucket, totalBuckets, true);
+ }
+
+ private WriterContainer<T> createWriterContainer(
+ BinaryRow partition, int bucket, int expectedTotalBuckets, boolean
validateNumBuckets) {
if (LOG.isDebugEnabled()) {
LOG.debug("Creating writer for partition {}, bucket {}",
partition, bucket);
}
@@ -496,7 +538,9 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
partition, bucket, latestSnapshot,
ignorePreviousFiles);
RestoreFiles restored = RestoreFiles.empty();
if (!actualIgnorePreviousFiles) {
- restored = scanExistingFileMetas(partition, bucket);
+ restored =
+ scanExistingFileMetas(
+ partition, bucket, expectedTotalBuckets,
validateNumBuckets);
}
DynamicBucketIndexMaintainer indexMaintainer =
@@ -541,7 +585,7 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
Snapshot previousSnapshot = restored.snapshot();
return new WriterContainer<>(
writer,
- firstNonNull(restored.totalBuckets(), numBuckets),
+ firstNonNull(restored.totalBuckets(), expectedTotalBuckets),
indexMaintainer,
dvMaintainer,
primaryKeyIndexMaintainer,
@@ -588,7 +632,8 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
return this;
}
- private RestoreFiles scanExistingFileMetas(BinaryRow partition, int
bucket) {
+ private RestoreFiles scanExistingFileMetas(
+ BinaryRow partition, int bucket, int expectedTotalBuckets, boolean
validateNumBuckets) {
Supplier<String> partInfo =
() ->
partitionType.getFieldCount() > 0
@@ -615,19 +660,33 @@ public abstract class AbstractFileStoreWrite<T>
implements FileStoreWrite<T> {
partInfo.get(), bucket),
e);
}
- Integer restoredTotalBuckets = restored.totalBuckets();
- int totalBuckets = numBuckets;
- if (restoredTotalBuckets != null) {
- totalBuckets = restoredTotalBuckets;
+ if (restored.totalBuckets() != null && validateNumBuckets) {
+ checkNumBuckets(partInfo.get(), expectedTotalBuckets,
restored.totalBuckets());
}
- if (!ignoreNumBucketCheck && totalBuckets != numBuckets) {
+ return restored;
+ }
+
+ private void checkNumBuckets(BinaryRow partition, int expected, int
previous) {
+ String partInfo =
+ partitionType.getFieldCount() > 0
+ ? "partition "
+ + getPartitionComputer(
+ partitionType,
+
PARTITION_DEFAULT_NAME.defaultValue(),
+ legacyPartitionName)
+ .generatePartValues(partition)
+ : "table";
+ checkNumBuckets(partInfo, expected, previous);
+ }
+
+ private void checkNumBuckets(String partInfo, int expected, int previous) {
+ if (expected != previous) {
throw new RuntimeException(
String.format(
"Try to write %s with a new bucket num %d, but the
previous bucket num is %d. "
+ "Please switch to batch mode, and
perform INSERT OVERWRITE to rescale current data layout first.",
- partInfo.get(), numBuckets, totalBuckets));
+ partInfo, expected, previous));
}
- return restored;
}
private ExecutorService compactExecutor() {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 5725dc89a7..e07fcb443f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -804,14 +804,14 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
return retryCount + 1;
}
- private void checkSameBucketFromSnapshot(
+ private void checkSameFixedBucketFromSnapshot(
List<ManifestEntry> deltaFiles, @Nullable Snapshot latestSnapshot)
{
if (latestSnapshot == null) {
return;
}
Map<BinaryRow, Integer> expectedTotalBuckets =
- conflictDetection.collectUncheckedBucketPartitions(deltaFiles);
+
conflictDetection.collectUncheckedFixedBucketPartitions(deltaFiles);
if (expectedTotalBuckets.isEmpty()) {
return;
}
@@ -820,14 +820,14 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
scanner.readTotalBuckets(
latestSnapshot, new
ArrayList<>(expectedTotalBuckets.keySet()));
Optional<RuntimeException> exception =
- conflictDetection.checkSameBucketByTotalBuckets(
+ conflictDetection.checkSameFixedBucketByTotalBuckets(
expectedTotalBuckets, previousTotalBuckets);
if (exception.isPresent()) {
throw exception.get();
}
}
- private boolean shouldCheckSameBucket(CommitKind commitKind) {
+ private boolean shouldCheckSameFixedBucket(CommitKind commitKind) {
return commitKind == CommitKind.APPEND
&& bucketMode == BucketMode.HASH_FIXED
&& (isUnorderedWriteOnlyAppend() ||
isWriteOnlySnapshotSequenceAppend());
@@ -940,6 +940,10 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
}
}
+ if (latestSnapshot == null) {
+ conflictDetection.checkSameBucketWithinDelta(deltaFiles);
+ }
+
List<BinaryRow> changedPartitions = null;
if (strictModeChecker != null) {
changedPartitions = changedPartitions(deltaFiles, indexFiles);
@@ -964,8 +968,8 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
boolean checkConflicts = latestSnapshot != null && (discardDuplicate
|| detectConflicts);
// By default, if checkConflicts is required, we do not have to do the
extra check bucket
// here.
- if (!checkConflicts && shouldCheckSameBucket(commitKind)) {
- checkSameBucketFromSnapshot(deltaFiles, latestSnapshot);
+ if (!checkConflicts && shouldCheckSameFixedBucket(commitKind)) {
+ checkSameFixedBucketFromSnapshot(deltaFiles, latestSnapshot);
}
if (checkConflicts) {
// latestSnapshotId is different from the snapshot id we've
checked for conflicts,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java
index b94bd04fc4..9f85047fb6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java
@@ -96,6 +96,16 @@ public interface FileStoreWrite<T> extends
Restorable<List<FileStoreWrite.State<
*/
void write(BinaryRow partition, int bucket, T data) throws Exception;
+ /**
+ * Write data with the total number of buckets explicitly determined at
runtime.
+ *
+ * <p>This is used when a partition's bucket count is not a static table
option. All writes to
+ * the same partition must use the same {@code totalBuckets}.
+ */
+ default void write(BinaryRow partition, int bucket, int totalBuckets, T
data) throws Exception {
+ throw new UnsupportedOperationException("Runtime bucket counts are not
supported.");
+ }
+
/**
* Compact data stored in given partition and bucket. Note that compaction
process is only
* submitted and may not be completed when the method returns.
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
index 34c5a29a46..4cdc5f997b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
@@ -75,7 +75,7 @@ import static
org.apache.paimon.utils.Preconditions.checkState;
public class ConflictDetection {
private static final Logger LOG =
LoggerFactory.getLogger(ConflictDetection.class);
- private static final int SAME_BUCKET_CHECK_CACHE_MAX_SIZE = 1000;
+ private static final int FIXED_BUCKET_CHECK_CACHE_MAX_SIZE = 1000;
private final String tableName;
private final String commitUser;
@@ -89,13 +89,15 @@ public class ConflictDetection {
private final IndexFileHandler indexFileHandler;
private final SnapshotManager snapshotManager;
private final CommitScanner commitScanner;
- private final Map<BinaryRow, Boolean> sameBucketCheckedPartitions =
- new LinkedHashMap<BinaryRow,
Boolean>(SAME_BUCKET_CHECK_CACHE_MAX_SIZE, 0.75f, false) {
- @Override
- protected boolean removeEldestEntry(Map.Entry<BinaryRow,
Boolean> eldest) {
- return size() > SAME_BUCKET_CHECK_CACHE_MAX_SIZE;
- }
- };
+ private final Set<BinaryRow> checkedFixedBucketPartitions =
+ Collections.newSetFromMap(
+ new LinkedHashMap<BinaryRow, Boolean>(
+ FIXED_BUCKET_CHECK_CACHE_MAX_SIZE, 0.75f, false) {
+ @Override
+ protected boolean
removeEldestEntry(Map.Entry<BinaryRow, Boolean> eldest) {
+ return size() > FIXED_BUCKET_CHECK_CACHE_MAX_SIZE;
+ }
+ });
private @Nullable PartitionExpire partitionExpire;
private @Nullable Long rowIdCheckFromSnapshot = null;
@@ -248,13 +250,22 @@ public class ConflictDetection {
latestSnapshot, deltaEntries, deltaIndexEntries,
rowIdColumnConflictChecker);
}
- public <T extends FileEntry> Map<BinaryRow, Integer>
collectUncheckedBucketPartitions(
+ public <T extends FileEntry> Map<BinaryRow, Integer>
collectUncheckedFixedBucketPartitions(
+ List<T> deltaEntries) {
+ Map<BinaryRow, Integer> totalBuckets =
collectBucketPartitions(deltaEntries);
+ totalBuckets.keySet().removeAll(checkedFixedBucketPartitions);
+ return totalBuckets;
+ }
+
+ public <T extends FileEntry> void checkSameBucketWithinDelta(List<T>
deltaEntries) {
+ collectBucketPartitions(deltaEntries);
+ }
+
+ private <T extends FileEntry> Map<BinaryRow, Integer>
collectBucketPartitions(
List<T> deltaEntries) {
Map<BinaryRow, Integer> totalBuckets = new HashMap<>();
for (T entry : deltaEntries) {
- if (entry.kind() != FileKind.ADD
- || entry.totalBuckets() <= 0
- ||
sameBucketCheckedPartitions.containsKey(entry.partition())) {
+ if (entry.kind() != FileKind.ADD || entry.totalBuckets() <= 0) {
continue;
}
@@ -266,7 +277,7 @@ public class ConflictDetection {
return totalBuckets;
}
- public Optional<RuntimeException> checkSameBucketByTotalBuckets(
+ public Optional<RuntimeException> checkSameFixedBucketByTotalBuckets(
Map<BinaryRow, Integer> expectedTotalBuckets,
Map<BinaryRow, Integer> previousTotalBuckets) {
for (Map.Entry<BinaryRow, Integer> entry :
expectedTotalBuckets.entrySet()) {
@@ -275,9 +286,7 @@ public class ConflictDetection {
return Optional.of(bucketNumMismatch(entry.getKey(),
entry.getValue(), previous));
}
}
- for (BinaryRow partition : expectedTotalBuckets.keySet()) {
- sameBucketCheckedPartitions.put(partition, Boolean.TRUE);
- }
+ checkedFixedBucketPartitions.addAll(expectedTotalBuckets.keySet());
return Optional.empty();
}
@@ -297,10 +306,6 @@ public class ConflictDetection {
if (entry.totalBuckets() <= 0) {
continue;
}
- if (sameBucketCheckedPartitions.containsKey(entry.partition())) {
- continue;
- }
-
if (!totalBuckets.containsKey(entry.partition())) {
totalBuckets.put(entry.partition(), entry.totalBuckets());
continue;
@@ -327,9 +332,6 @@ public class ConflictDetection {
LOG.warn("", conflictException.getLeft());
return Optional.of(conflictException.getRight());
}
- for (BinaryRow partition : totalBuckets.keySet()) {
- sameBucketCheckedPartitions.put(partition, Boolean.TRUE);
- }
return Optional.empty();
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java
index e1064e7045..36e8d03340 100644
---
a/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java
@@ -264,6 +264,13 @@ public class PrivilegedFileStoreTable extends
DelegatedFileStoreTable {
return wrapped.newWrite(commitUser, writeId);
}
+ @Override
+ public TableWriteImpl<?> newPostponeFixedBucketWrite(
+ String commitUser, @Nullable Integer writeId) {
+ privilegeChecker.assertCanInsert(identifier);
+ return wrapped.newPostponeFixedBucketWrite(commitUser, writeId);
+ }
+
@Override
public TableCommitImpl newCommit(String commitUser) {
privilegeChecker.assertCanInsert(identifier);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java
index f9bcbeee01..9a5e81bcb6 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java
@@ -345,6 +345,12 @@ public abstract class DelegatedFileStoreTable implements
FileStoreTable {
return wrapped.newWrite(commitUser, writeId);
}
+ @Override
+ public TableWriteImpl<?> newPostponeFixedBucketWrite(
+ String commitUser, @Nullable Integer writeId) {
+ return wrapped.newPostponeFixedBucketWrite(commitUser, writeId);
+ }
+
@Override
public TableCommitImpl newCommit(String commitUser) {
return wrapped.newCommit(commitUser);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java
index 5a3f87d0ed..9704934305 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java
@@ -29,6 +29,7 @@ import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.query.LocalTableQuery;
import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.PostponeFixedBucketWriteBuilder;
import org.apache.paimon.table.sink.RowKeyExtractor;
import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.table.sink.TableWriteImpl;
@@ -131,8 +132,20 @@ public interface FileStoreTable extends DataTable {
@Override
TableWriteImpl<?> newWrite(String commitUser);
+ /** Returns a builder for fixed-bucket batch writes to a postpone-bucket
table. */
+ default PostponeFixedBucketWriteBuilder
newPostponeFixedBucketWriteBuilder() {
+ return new PostponeFixedBucketWriteBuilder(this);
+ }
+
TableWriteImpl<?> newWrite(String commitUser, @Nullable Integer writeId);
+ /** Creates a fixed-bucket merge-tree write for a postpone-bucket batch
write. */
+ default TableWriteImpl<?> newPostponeFixedBucketWrite(
+ String commitUser, @Nullable Integer writeId) {
+ throw new UnsupportedOperationException(
+ "Postpone fixed-bucket writes are only supported by
primary-key tables.");
+ }
+
@Override
TableCommitImpl newCommit(String commitUser);
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 5335830988..60b464fd15 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
@@ -341,15 +341,6 @@ public class PostponeUtils {
return rowCounts;
}
- public static FileStoreTable tableForFixBucketWrite(FileStoreTable table) {
- Map<String, String> batchWriteOptions = new HashMap<>();
- batchWriteOptions.put(WRITE_ONLY.key(), "true");
- // It's just used to create merge tree writer for writing files to
fixed bucket.
- // The real bucket number is determined at runtime.
- batchWriteOptions.put(BUCKET.key(), "1");
- return table.copy(batchWriteOptions);
- }
-
public static FileStoreTable tableForPostponeCompact(
FileStoreTable table, int numBuckets, long snapshotId) {
Map<String, String> compactOptions = new HashMap<>();
@@ -359,11 +350,6 @@ public class PostponeUtils {
return table.copy(compactOptions);
}
- public static FileStoreTable tableForCommit(FileStoreTable table) {
- return table.copy(
- Collections.singletonMap(BUCKET.key(),
String.valueOf(BucketMode.POSTPONE_BUCKET)));
- }
-
/** Snapshot-bound bucket-count assignment. */
public static final class PostponeBucketAssigner implements Serializable {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
index 3030b3504e..5e48557d76 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
@@ -26,6 +26,7 @@ import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.mergetree.compact.LookupMergeFunction;
import org.apache.paimon.mergetree.compact.MergeFunctionFactory;
+import org.apache.paimon.operation.AbstractFileStoreWrite;
import org.apache.paimon.operation.FileStoreScan;
import org.apache.paimon.operation.KeyValueFileStoreScan;
import org.apache.paimon.predicate.Predicate;
@@ -173,10 +174,20 @@ public class PrimaryKeyFileStoreTable extends
AbstractFileStoreTable {
@Override
public TableWriteImpl<KeyValue> newWrite(String commitUser, @Nullable
Integer writeId) {
+ return newWrite(store().newWrite(commitUser, writeId));
+ }
+
+ @Override
+ public TableWriteImpl<KeyValue> newPostponeFixedBucketWrite(
+ String commitUser, @Nullable Integer writeId) {
+ return newWrite(store().newPostponeFixedBucketWrite(commitUser));
+ }
+
+ private TableWriteImpl<KeyValue> newWrite(AbstractFileStoreWrite<KeyValue>
storeWrite) {
KeyValue kv = new KeyValue();
return new TableWriteImpl<>(
rowType(),
- store().newWrite(commitUser, writeId),
+ storeWrite,
createRowKeyExtractor(),
(record, rowKind) ->
kv.replace(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
index c06bd9de42..d8c97405e2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
@@ -21,7 +21,6 @@ package org.apache.paimon.table.sink;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.InnerTable;
-import org.apache.paimon.table.Table;
import org.apache.paimon.types.RowType;
import javax.annotation.Nullable;
@@ -40,7 +39,6 @@ public class BatchWriteBuilderImpl implements
BatchWriteBuilder {
private final String commitUser;
private Map<String, String> staticPartition;
- private boolean appendCommitCheckConflict = false;
private @Nullable Long rowIdCheckFromSnapshot = null;
public BatchWriteBuilderImpl(InnerTable table) {
@@ -48,13 +46,6 @@ public class BatchWriteBuilderImpl implements
BatchWriteBuilder {
this.commitUser = createCommitUser(new Options(table.options()));
}
- private BatchWriteBuilderImpl(
- InnerTable table, String commitUser, @Nullable Map<String, String>
staticPartition) {
- this.table = table;
- this.commitUser = commitUser;
- this.staticPartition = staticPartition;
- }
-
@Override
public String tableName() {
return table.name();
@@ -86,7 +77,6 @@ public class BatchWriteBuilderImpl implements
BatchWriteBuilder {
InnerTableCommit commit =
table.newCommit(commitUser)
.withOverwrite(staticPartition)
- .appendCommitCheckConflict(appendCommitCheckConflict)
.rowIdCheckConflict(rowIdCheckFromSnapshot);
commit.ignoreEmptyCommit(
Options.fromMap(table.options())
@@ -95,15 +85,6 @@ public class BatchWriteBuilderImpl implements
BatchWriteBuilder {
return commit;
}
- public BatchWriteBuilderImpl copyWithNewTable(Table newTable) {
- return new BatchWriteBuilderImpl((InnerTable) newTable, commitUser,
staticPartition);
- }
-
- public BatchWriteBuilderImpl appendCommitCheckConflict(boolean
appendCommitCheckConflict) {
- this.appendCommitCheckConflict = appendCommitCheckConflict;
- return this;
- }
-
public BatchWriteBuilderImpl rowIdCheckConflict(@Nullable Long
rowIdCheckFromSnapshot) {
this.rowIdCheckFromSnapshot = rowIdCheckFromSnapshot;
return this;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java
similarity index 50%
copy from
paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
copy to
paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java
index c06bd9de42..a3b28e47cc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java
@@ -20,8 +20,8 @@ package org.apache.paimon.table.sink;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.options.Options;
-import org.apache.paimon.table.InnerTable;
-import org.apache.paimon.table.Table;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.RowType;
import javax.annotation.Nullable;
@@ -30,31 +30,26 @@ import java.util.Map;
import java.util.Optional;
import static org.apache.paimon.CoreOptions.createCommitUser;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Implementation for {@link WriteBuilder}. */
-public class BatchWriteBuilderImpl implements BatchWriteBuilder {
+/** Write builder for assigning fixed buckets at runtime to a postpone-bucket
table. */
+public class PostponeFixedBucketWriteBuilder implements BatchWriteBuilder {
private static final long serialVersionUID = 1L;
- private final InnerTable table;
+ private final FileStoreTable table;
private final String commitUser;
- private Map<String, String> staticPartition;
- private boolean appendCommitCheckConflict = false;
- private @Nullable Long rowIdCheckFromSnapshot = null;
+ @Nullable private Map<String, String> staticPartition;
- public BatchWriteBuilderImpl(InnerTable table) {
+ public PostponeFixedBucketWriteBuilder(FileStoreTable table) {
+ checkArgument(
+ table.bucketMode() == BucketMode.POSTPONE_MODE,
+ "Postpone fixed-bucket write requires a postpone-bucket
table.");
this.table = table;
this.commitUser = createCommitUser(new Options(table.options()));
}
- private BatchWriteBuilderImpl(
- InnerTable table, String commitUser, @Nullable Map<String, String>
staticPartition) {
- this.table = table;
- this.commitUser = commitUser;
- this.staticPartition = staticPartition;
- }
-
@Override
public String tableName() {
return table.name();
@@ -71,41 +66,34 @@ public class BatchWriteBuilderImpl implements
BatchWriteBuilder {
}
@Override
- public BatchWriteBuilder withOverwrite(@Nullable Map<String, String>
staticPartition) {
+ public PostponeFixedBucketWriteBuilder withOverwrite(
+ @Nullable Map<String, String> staticPartition) {
this.staticPartition = staticPartition;
return this;
}
@Override
- public BatchTableWrite newWrite() {
- return
table.newWrite(commitUser).withIgnorePreviousFiles(staticPartition != null);
+ public TableWriteImpl<?> newWrite() {
+ return newWrite(commitUser,
null).withIgnorePreviousFiles(staticPartition != null);
+ }
+
+ public TableWriteImpl<?> newWrite(String commitUser, @Nullable Integer
writeId) {
+ return table.newPostponeFixedBucketWrite(commitUser, writeId);
}
@Override
- public BatchTableCommit newCommit() {
- InnerTableCommit commit =
- table.newCommit(commitUser)
- .withOverwrite(staticPartition)
- .appendCommitCheckConflict(appendCommitCheckConflict)
- .rowIdCheckConflict(rowIdCheckFromSnapshot);
- commit.ignoreEmptyCommit(
+ public TableCommitImpl newCommit() {
+ boolean ignoreEmpty =
Options.fromMap(table.options())
.getOptional(CoreOptions.SNAPSHOT_IGNORE_EMPTY_COMMIT)
- .orElse(true));
- return commit;
+ .orElse(true);
+ return newCommit(commitUser, ignoreEmpty);
}
- public BatchWriteBuilderImpl copyWithNewTable(Table newTable) {
- return new BatchWriteBuilderImpl((InnerTable) newTable, commitUser,
staticPartition);
- }
-
- public BatchWriteBuilderImpl appendCommitCheckConflict(boolean
appendCommitCheckConflict) {
- this.appendCommitCheckConflict = appendCommitCheckConflict;
- return this;
- }
-
- public BatchWriteBuilderImpl rowIdCheckConflict(@Nullable Long
rowIdCheckFromSnapshot) {
- this.rowIdCheckFromSnapshot = rowIdCheckFromSnapshot;
- return this;
+ public TableCommitImpl newCommit(String commitUser, boolean
ignoreEmptyCommit) {
+ return table.newCommit(commitUser)
+ .withOverwrite(staticPartition)
+ .appendCommitCheckConflict(true)
+ .ignoreEmptyCommit(ignoreEmptyCommit);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
index 101a51e8a5..9c815db10d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
@@ -181,6 +181,21 @@ public class TableWriteImpl<T> implements InnerTableWrite,
Restorable<List<State
@Nullable
public SinkRecord writeAndReturn(InternalRow row, int bucket) throws
Exception {
+ return writeAndReturn(row, bucket, null);
+ }
+
+ /**
+ * Write a row to a bucket whose partition-level total bucket count is
determined at runtime.
+ */
+ @Nullable
+ public SinkRecord writeAndReturn(InternalRow row, int bucket, int
totalBuckets)
+ throws Exception {
+ return writeAndReturn(row, bucket, Integer.valueOf(totalBuckets));
+ }
+
+ @Nullable
+ private SinkRecord writeAndReturn(InternalRow row, int bucket, @Nullable
Integer totalBuckets)
+ throws Exception {
checkNullability(row);
row = wrapDefaultValue(row);
RowKind rowKind = RowKindGenerator.getRowKind(rowKindGenerator, row);
@@ -188,7 +203,12 @@ public class TableWriteImpl<T> implements InnerTableWrite,
Restorable<List<State
return null;
}
SinkRecord record = bucket == -1 ? toSinkRecord(row) :
toSinkRecord(row, bucket);
- write.write(record.partition(), record.bucket(),
recordExtractor.extract(record, rowKind));
+ T extracted = recordExtractor.extract(record, rowKind);
+ if (totalBuckets == null) {
+ write.write(record.partition(), record.bucket(), extracted);
+ } else {
+ write.write(record.partition(), record.bucket(), totalBuckets,
extracted);
+ }
return record;
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index 98606bfb10..fb22df8c1f 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -107,6 +107,7 @@ import java.util.stream.Collectors;
import static org.apache.paimon.index.HashIndexFile.HASH_INDEX;
import static
org.apache.paimon.partition.PartitionPredicate.createPartitionPredicate;
import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
+import static org.apache.paimon.table.BucketMode.POSTPONE_BUCKET;
import static
org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
import static org.apache.paimon.utils.HintFileUtils.LATEST;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -259,6 +260,69 @@ public class FileStoreCommitTest {
}
}
+ @Test
+ public void testPostponeBucketCheckIsNotSkippedByCache() throws Exception {
+ TestFileStore store = createStore(false, POSTPONE_BUCKET);
+ BinaryRow partition =
+ gen.getPartition(gen.nextInsert("20201110", 10, 1L, new int[]
{1, 1}, "first"));
+
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ assertThat(
+ commit.tryCommitOnce(
+ null,
+
Collections.singletonList(addFile(partition, 0, 2, 0)),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ 0,
+ null,
+ new HashMap<>(),
+ Snapshot.CommitKind.APPEND,
+ false,
+ null,
+ true,
+ null)
+ .isSuccess())
+ .isTrue();
+
+ Snapshot latestSnapshot = store.snapshotManager().latestSnapshot();
+ assertThat(
+ commit.tryCommitOnce(
+ null,
+
Collections.singletonList(addFile(partition, 1, 2, 1)),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ 1,
+ null,
+ new HashMap<>(),
+ Snapshot.CommitKind.APPEND,
+ false,
+ latestSnapshot,
+ true,
+ null)
+ .isSuccess())
+ .isTrue();
+
+ latestSnapshot = store.snapshotManager().latestSnapshot();
+ Snapshot finalLatestSnapshot = latestSnapshot;
+ assertThatThrownBy(
+ () ->
+ commit.tryCommitOnce(
+ null,
+
Collections.singletonList(addFile(partition, 2, 3, 2)),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ 2,
+ null,
+ new HashMap<>(),
+ Snapshot.CommitKind.APPEND,
+ false,
+ finalLatestSnapshot,
+ true,
+ null))
+ .hasMessageContaining("changed from 2 to 3 without
overwrite");
+ }
+ }
+
@Test
public void testFilterCommittedAfterExpire() throws Exception {
testRandomConcurrentNoConflict(1, false,
CoreOptions.ChangelogProducer.NONE);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
index 4bda9faf3e..3ea3628fa5 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
@@ -60,6 +60,7 @@ import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.InnerTableCommit;
+import org.apache.paimon.table.sink.PostponeFixedBucketWriteBuilder;
import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.sink.StreamTableWrite;
import org.apache.paimon.table.sink.StreamWriteBuilder;
@@ -212,6 +213,60 @@ public class PrimaryKeySimpleTableTest extends
SimpleTableTestBase {
assertThat(file.valueStatsCols()).isEmpty();
}
+ @Test
+ public void testPostponeFixedBucketWriteBuilder() throws Exception {
+ FileStoreTable table =
+ createFileStoreTable(options -> options.set(BUCKET,
BucketMode.POSTPONE_BUCKET));
+ PostponeFixedBucketWriteBuilder builder =
table.newPostponeFixedBucketWriteBuilder();
+
+ List<CommitMessage> conflictingMessages = new ArrayList<>();
+ try (TableWriteImpl<?> write = builder.newWrite()) {
+ write.writeAndReturn(rowData(1, 1, 1L), 0, 2);
+ conflictingMessages.addAll(write.prepareCommit());
+ }
+ try (TableWriteImpl<?> write = builder.newWrite()) {
+ write.writeAndReturn(rowData(1, 2, 2L), 1, 3);
+ conflictingMessages.addAll(write.prepareCommit());
+ }
+ try (BatchTableCommit commit = builder.newCommit()) {
+ assertThatThrownBy(() -> commit.commit(conflictingMessages))
+ .hasMessageContaining("new bucket num 3")
+ .hasMessageContaining("previous bucket num is 2");
+ }
+ assertThat(table.latestSnapshot()).isEmpty();
+
+ List<CommitMessage> messages;
+ try (TableWriteImpl<?> write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ assertThatThrownBy(() -> write.writeAndReturn(rowData(1, 1, 1L),
0, 0))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be positive");
+ assertThatThrownBy(() -> write.writeAndReturn(rowData(1, 1, 1L),
2, 2))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("out of range");
+ write.writeAndReturn(rowData(1, 1, 1L), 0, 2);
+ assertThatThrownBy(() -> write.writeAndReturn(rowData(1, 3, 3L),
1, 3))
+ .hasMessageContaining("new bucket num 3")
+ .hasMessageContaining("previous bucket num is 2");
+ write.writeAndReturn(rowData(2, 2, 2L), 2, 3);
+ messages = write.prepareCommit();
+ assertThat(messages)
+ .extracting(message -> ((CommitMessageImpl)
message).totalBuckets())
+ .containsExactlyInAnyOrder(2, 3);
+ commit.commit(messages);
+ }
+
+ assertThat(PostponeUtils.getKnownNumBuckets(table))
+ .containsEntry(binaryRow(1), 2)
+ .containsEntry(binaryRow(2), 3);
+
+ try (TableWriteImpl<?> write = builder.newWrite()) {
+ assertThatThrownBy(() -> write.writeAndReturn(rowData(1, 3, 3L),
0, 3))
+ .hasMessageContaining("new bucket num 3")
+ .hasMessageContaining("previous bucket num is 2");
+ }
+ }
+
@ParameterizedTest(name = "format-{0}")
@ValueSource(strings = {"avro", "parquet"})
public void testStatsModePerLevel(String format) throws Exception {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
index b323301c79..5d00584606 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -136,14 +136,10 @@ public abstract class FlinkSink<T> implements
Serializable {
StreamExecutionEnvironment env = input.getExecutionEnvironment();
boolean isStreaming = isStreaming(input);
- boolean writeOnly = table.coreOptions().writeOnly();
+ boolean writeOnly = writeOnly();
StoreSinkWrite.Provider writeProvider =
- StoreSinkWrite.createWriteProvider(
- table,
- env.getCheckpointConfig(),
- isStreaming,
- ignorePreviousFiles,
- hasSinkMaterializer(input));
+ createWriteProvider(
+ env.getCheckpointConfig(), isStreaming,
hasSinkMaterializer(input));
writeProvider =
StoreSinkWrite.withBlobDescriptorReaderFactory(
writeProvider, blobDescriptorReaderFactory);
@@ -199,6 +195,20 @@ public abstract class FlinkSink<T> implements Serializable
{
return written;
}
+ protected boolean writeOnly() {
+ return table.coreOptions().writeOnly();
+ }
+
+ protected StoreSinkWrite.Provider createWriteProvider(
+ CheckpointConfig checkpointConfig, boolean isStreaming, boolean
hasSinkMaterializer) {
+ return StoreSinkWrite.createWriteProvider(
+ table, checkpointConfig, isStreaming, ignorePreviousFiles,
hasSinkMaterializer);
+ }
+
+ protected boolean ignorePreviousFiles() {
+ return ignorePreviousFiles;
+ }
+
public DataStreamSink<?> doCommit(DataStream<Committable> written, String
commitUser) {
StreamExecutionEnvironment env = written.getExecutionEnvironment();
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
index 55ed3cf3a0..656a68db6e 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
@@ -368,11 +368,8 @@ public class FlinkSinkBuilder {
new
PostponeFixedBucketChannelComputer(table.schema(), knownNumBuckets),
parallelism);
- FileStoreTable tableForWrite =
PostponeUtils.tableForFixBucketWrite(table);
-
return configureBlobDescriptorReaderFactory(
- new PostponeFixedBucketSink(
- tableForWrite, overwritePartition,
knownNumBuckets))
+ new PostponeFixedBucketSink(table,
overwritePartition, knownNumBuckets))
.sinkFrom(partitioned);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeBatchWriteOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeBatchWriteOperator.java
index 5fc50542f0..a9438e1033 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeBatchWriteOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeBatchWriteOperator.java
@@ -27,7 +27,6 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.utils.RuntimeContextUtils;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.RowPartitionKeyExtractor;
import org.apache.paimon.table.sink.SinkRecord;
@@ -35,14 +34,9 @@ import
org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import javax.annotation.Nullable;
-import java.io.IOException;
-import java.util.ArrayList;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
-import static org.apache.paimon.utils.Preconditions.checkNotNull;
-
/** Writer to write {@link InternalRow} to postpone fixed bucket. */
public class PostponeBatchWriteOperator extends
StatelessRowDataStoreWriteOperator {
@@ -83,7 +77,6 @@ public class PostponeBatchWriteOperator extends
StatelessRowDataStoreWriteOperat
this.bucketKeyProjection =
CodeGenUtils.newProjection(
schema.logicalRowType(),
schema.projection(schema.bucketKeys()));
- ((StoreSinkWriteImpl)
write).getWrite().getWrite().withIgnoreNumBucketCheck(true);
}
@Override
@@ -92,26 +85,6 @@ public class PostponeBatchWriteOperator extends
StatelessRowDataStoreWriteOperat
BinaryRow partition = partitionKeyExtractor.partition(row);
int numBuckets = knownNumBuckets.computeIfAbsent(partition.copy(), p
-> defaultNumBuckets);
int bucket = bucketFunction.bucket(bucketKeyProjection.apply(row),
numBuckets);
- return write.write(row, bucket);
- }
-
- @Override
- protected List<Committable> prepareCommit(boolean waitCompaction, long
checkpointId)
- throws IOException {
- List<Committable> committables = new ArrayList<>();
- for (Committable committable : super.prepareCommit(waitCompaction,
checkpointId)) {
- CommitMessageImpl message = (CommitMessageImpl)
committable.commitMessage();
- committables.add(
- new Committable(
- committable.checkpointId(),
- new CommitMessageImpl(
- message.partition(),
- message.bucket(),
-
checkNotNull(knownNumBuckets.get(message.partition())),
- message.newFilesIncrement(),
- message.compactIncrement())));
- }
-
- return committables;
+ return write.write(row, bucket, numBuckets);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeFixedBucketChannelComputer.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeFixedBucketChannelComputer.java
index e1bf08d843..8e1c7325f2 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeFixedBucketChannelComputer.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeFixedBucketChannelComputer.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.sink;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.schema.TableSchema;
@@ -36,6 +37,7 @@ public class PostponeFixedBucketChannelComputer implements
ChannelComputer<Inter
private final TableSchema schema;
private final Map<BinaryRow, Integer> knownNumBuckets;
+ private final int maxNumBuckets;
private transient int numChannels;
private transient FixedBucketRowKeyExtractor keyExtractor;
@@ -44,6 +46,8 @@ public class PostponeFixedBucketChannelComputer implements
ChannelComputer<Inter
TableSchema schema, Map<BinaryRow, Integer> knownNumBuckets) {
this.schema = schema;
this.knownNumBuckets = knownNumBuckets;
+ this.maxNumBuckets =
+ new
CoreOptions(schema.options()).postponeBatchWriteFixedBucketMaxParallelism();
}
@Override
@@ -56,7 +60,9 @@ public class PostponeFixedBucketChannelComputer implements
ChannelComputer<Inter
public int channel(InternalRow record) {
keyExtractor.setRecord(record);
BinaryRow partition = keyExtractor.partition();
- int numBuckets = knownNumBuckets.computeIfAbsent(partition, p ->
numChannels);
+ int numBuckets =
+ knownNumBuckets.computeIfAbsent(
+ partition.copy(), p -> Math.min(numChannels,
maxNumBuckets));
int bucket = keyExtractor.bucket(numBuckets);
return ChannelComputer.select(partition, bucket, numChannels);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeFixedBucketSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeFixedBucketSink.java
index 130804f73c..ad7d78cdff 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeFixedBucketSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeFixedBucketSink.java
@@ -22,8 +22,8 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.PostponeUtils;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
@@ -65,31 +65,28 @@ public class PostponeFixedBucketSink extends
FlinkWriteSink<InternalRow> {
return createRestoreOnlyCommittableStateManager(table);
}
+ @Override
+ protected boolean writeOnly() {
+ return true;
+ }
+
+ @Override
+ protected StoreSinkWrite.Provider createWriteProvider(
+ CheckpointConfig checkpointConfig, boolean isStreaming, boolean
hasSinkMaterializer) {
+ return StoreSinkWrite.createPostponeFixedBucketWriteProvider(
+ ignorePreviousFiles(), isStreaming, hasSinkMaterializer);
+ }
+
@Override
protected Committer.Factory<Committable, ManifestCommittable>
createCommitterFactory() {
- if (overwritePartition == null) {
- // The table has copied bucket option outside, no need to change.
- return context ->
- new StoreCommitter(
- table,
- table.newCommit(context.commitUser())
- .withOverwrite(overwritePartition)
-
.ignoreEmptyCommit(!context.streamingCheckpointEnabled())
- // Need to check conflict
- .appendCommitCheckConflict(true),
- context);
- } else {
- // When overwriting, the postpone bucket files need to be deleted,
so using a postpone
- // bucket table commit here
- FileStoreTable tableForCommit =
PostponeUtils.tableForCommit(table);
- return context ->
- new StoreCommitter(
- tableForCommit,
- tableForCommit
- .newCommit(context.commitUser())
- .withOverwrite(overwritePartition)
-
.ignoreEmptyCommit(!context.streamingCheckpointEnabled()),
- context);
- }
+ return context ->
+ new StoreCommitter(
+ table,
+ table.newPostponeFixedBucketWriteBuilder()
+ .withOverwrite(overwritePartition)
+ .newCommit(
+ context.commitUser(),
+ !context.streamingCheckpointEnabled()),
+ context);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWrite.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWrite.java
index 5aee320f6e..474a996db6 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWrite.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWrite.java
@@ -59,6 +59,9 @@ public interface StoreSinkWrite {
@Nullable
SinkRecord write(InternalRow rowData, int bucket) throws Exception;
+ @Nullable
+ SinkRecord write(InternalRow rowData, int bucket, int totalBuckets) throws
Exception;
+
void compact(BinaryRow partition, int bucket, boolean fullCompaction)
throws Exception;
void notifyNewFiles(long snapshotId, BinaryRow partition, int bucket,
List<DataFileMeta> files);
@@ -203,4 +206,29 @@ public interface StoreSinkWrite {
metricGroup);
};
}
+
+ static StoreSinkWrite.Provider createPostponeFixedBucketWriteProvider(
+ boolean ignorePreviousFiles, boolean isStreaming, boolean
hasSinkMaterializer) {
+ return (table, commitUser, state, ioManager, memoryPoolFactory,
metricGroup) -> {
+ Preconditions.checkArgument(
+ !hasSinkMaterializer,
+ String.format(
+ "Sink materializer must not be used with Paimon
sink. "
+ + "Please set '%s' to '%s' in Flink's
config.",
+
ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE.key(),
+
ExecutionConfigOptions.UpsertMaterialize.NONE.name()));
+ return new StoreSinkWriteImpl(
+ table,
+ commitUser,
+ state,
+ ioManager,
+ ignorePreviousFiles,
+ false,
+ isStreaming,
+ memoryPoolFactory,
+ metricGroup,
+ (t, user, writeId) ->
+
t.newPostponeFixedBucketWriteBuilder().newWrite(user, writeId));
+ };
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java
index e314b997c5..797f6ad86d 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java
@@ -57,6 +57,7 @@ public class StoreSinkWriteImpl implements StoreSinkWrite {
private final boolean isStreamingMode;
private final MemoryPoolFactory memoryPoolFactory;
@Nullable private final MetricGroup metricGroup;
+ private final TableWriteFactory tableWriteFactory;
@Nullable private UriReaderFactory blobDescriptorReaderFactory;
@@ -72,6 +73,30 @@ public class StoreSinkWriteImpl implements StoreSinkWrite {
boolean isStreamingMode,
MemoryPoolFactory memoryPoolFactory,
@Nullable MetricGroup metricGroup) {
+ this(
+ table,
+ commitUser,
+ state,
+ ioManager,
+ ignorePreviousFiles,
+ waitCompaction,
+ isStreamingMode,
+ memoryPoolFactory,
+ metricGroup,
+ FileStoreTable::newWrite);
+ }
+
+ StoreSinkWriteImpl(
+ FileStoreTable table,
+ String commitUser,
+ StoreSinkWriteState state,
+ IOManager ioManager,
+ boolean ignorePreviousFiles,
+ boolean waitCompaction,
+ boolean isStreamingMode,
+ MemoryPoolFactory memoryPoolFactory,
+ @Nullable MetricGroup metricGroup,
+ TableWriteFactory tableWriteFactory) {
this.commitUser = commitUser;
this.state = state;
this.paimonIOManager = new
IOManagerImpl(ioManager.getSpillingDirectoriesPaths());
@@ -80,12 +105,14 @@ public class StoreSinkWriteImpl implements StoreSinkWrite {
this.isStreamingMode = isStreamingMode;
this.memoryPoolFactory = memoryPoolFactory;
this.metricGroup = metricGroup;
+ this.tableWriteFactory = tableWriteFactory;
this.write = newTableWrite(table);
}
private TableWriteImpl<?> newTableWrite(FileStoreTable table) {
TableWriteImpl<?> tableWrite =
- table.newWrite(commitUser, state.getSubtaskId())
+ tableWriteFactory
+ .create(table, commitUser, state.getSubtaskId())
.withIOManager(paimonIOManager)
.withIgnorePreviousFiles(ignorePreviousFiles)
.withMemoryPoolFactory(memoryPoolFactory);
@@ -122,6 +149,12 @@ public class StoreSinkWriteImpl implements StoreSinkWrite {
return write.writeAndReturn(withBlobDescriptorReader(rowData), bucket);
}
+ @Override
+ @Nullable
+ public SinkRecord write(InternalRow rowData, int bucket, int totalBuckets)
throws Exception {
+ return write.writeAndReturn(withBlobDescriptorReader(rowData), bucket,
totalBuckets);
+ }
+
private InternalRow withBlobDescriptorReader(InternalRow rowData) {
return blobDescriptorReaderFactory == null
? rowData
@@ -198,4 +231,11 @@ public class StoreSinkWriteImpl implements StoreSinkWrite {
public TableWriteImpl<?> getWrite() {
return write;
}
+
+ @FunctionalInterface
+ interface TableWriteFactory {
+
+ TableWriteImpl<?> create(
+ FileStoreTable table, String commitUser, @Nullable Integer
writeId);
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/PostponeFixedBucketChannelComputerTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/PostponeFixedBucketChannelComputerTest.java
index 991ab5d2dd..be05255938 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/PostponeFixedBucketChannelComputerTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/PostponeFixedBucketChannelComputerTest.java
@@ -166,4 +166,34 @@ public class PostponeFixedBucketChannelComputerTest {
assertThat(computer.channel(row1)).isEqualTo(computer.channel(row2));
}
}
+
+ @Test
+ public void testDefaultBucketNumberRespectsMaxParallelism() throws
Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.BIGINT()},
+ new String[] {"pt", "k"});
+
+ SchemaManager schemaManager =
+ new SchemaManager(LocalFileIO.create(), new
Path(tempDir.toString()));
+ Map<String, String> options = new HashMap<>();
+ options.put("bucket", "-2");
+ options.put("postpone.batch-write-fixed-bucket.max-parallelism", "2");
+ TableSchema schema =
+ schemaManager.createTable(
+ new Schema(
+ rowType.getFields(),
+ Collections.singletonList("pt"),
+ Arrays.asList("pt", "k"),
+ options,
+ ""));
+
+ Map<BinaryRow, Integer> knownNumBuckets = new HashMap<>();
+ PostponeFixedBucketChannelComputer computer =
+ new PostponeFixedBucketChannelComputer(schema,
knownNumBuckets);
+ computer.setup(8);
+ computer.channel(GenericRow.of(1, 1L));
+
+ assertThat(knownNumBuckets).containsValue(2);
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreCompactOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreCompactOperatorTest.java
index 5e2cb10e0e..cf78c43192 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreCompactOperatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreCompactOperatorTest.java
@@ -326,6 +326,11 @@ public class StoreCompactOperatorTest extends
TableTestBase {
return null;
}
+ @Override
+ public SinkRecord write(InternalRow rowData, int bucket, int
totalBuckets) {
+ return null;
+ }
+
@Override
public void compact(BinaryRow partition, int bucket, boolean
fullCompaction) {
compactTime++;
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 aaba0f1ca0..e1301703da 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
@@ -88,12 +88,11 @@ case class PaimonSparkWriter(
table.bucketMode() == POSTPONE_MODE &&
coreOptions.postponeBatchWriteFixedBucket()
val writeBuilder: BatchWriteBuilder = {
- val tableForWrite = if (postponeBatchWriteFixedBucket) {
- PostponeUtils.tableForFixBucketWrite(table)
+ if (postponeBatchWriteFixedBucket) {
+ table.newPostponeFixedBucketWriteBuilder()
} else {
- table
+ table.newBatchWriteBuilder()
}
- tableForWrite.newBatchWriteBuilder()
}
def writeOnly(): PaimonSparkWriter = {
@@ -438,16 +437,7 @@ case class PaimonSparkWriter(
}
def commit(commitMessages: Seq[CommitMessage], operation:
Snapshot.Operation): Unit = {
- val finalWriteBuilder = if (postponeBatchWriteFixedBucket) {
- writeBuilder
- .asInstanceOf[BatchWriteBuilderImpl]
- .copyWithNewTable(PostponeUtils.tableForCommit(table))
- // Need to check conflict
- .appendCommitCheckConflict(true)
- } else {
- writeBuilder
- }
- val tableCommit = finalWriteBuilder.newCommit()
+ val tableCommit = writeBuilder.newCommit()
if (operation != null) {
tableCommit.withOperation(operation)
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala
index a954bd64b2..af20144f52 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala
@@ -50,9 +50,6 @@ case class PaimonDataWrite(
if (writeRowTracking) {
_write.withWriteType(writeType)
}
- if (postponePartitionBucketComputer.isDefined) {
- _write.getWrite.withIgnoreNumBucketCheck(true)
- }
_write
}
@@ -65,27 +62,17 @@ case class PaimonDataWrite(
}
def write(row: Row, bucket: Int): Unit = {
- postWrite(write.writeAndReturn(toPaimonRow(row), bucket))
+ val paimonRow = toPaimonRow(row)
+ val sinkRecord = postponePartitionBucketComputer match {
+ case Some(numBuckets) =>
+ write.writeAndReturn(paimonRow, bucket,
numBuckets(write.getPartition(paimonRow)))
+ case None => write.writeAndReturn(paimonRow, bucket)
+ }
+ postWrite(sinkRecord)
}
override def commitImpl(): Seq[CommitMessage] = {
- val commitMessages = write.prepareCommit().asScala.toSeq
-
- if (postponePartitionBucketComputer.isDefined) {
- commitMessages.map {
- case message: CommitMessageImpl =>
- new CommitMessageImpl(
- message.partition(),
- message.bucket(),
- postponePartitionBucketComputer.get.apply(message.partition()),
- message.newFilesIncrement(),
- message.compactIncrement()
- )
- case _ => throw new RuntimeException()
- }
- } else {
- commitMessages
- }
+ write.prepareCommit().asScala.toSeq
}
override def close(): Unit = {
