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 f7ec93b3cc [flink] SortedIndexTopoBuilder supports custom commit-user
(#8646)
f7ec93b3cc is described below
commit f7ec93b3cc1edeccd10ac23399c0fbcd3569676c
Author: yuzelin <[email protected]>
AuthorDate: Wed Jul 15 18:45:08 2026 +0800
[flink] SortedIndexTopoBuilder supports custom commit-user (#8646)
---
.../apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
index 2b627913ef..d9f63cf64a 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
@@ -76,7 +76,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.UUID;
import java.util.function.Supplier;
import static
org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder.groupSplitsByRange;
@@ -203,7 +202,7 @@ public class SortedIndexTopoBuilder {
@SuppressWarnings("unchecked")
DataStream<Committable>[] rest =
allStreams.subList(1, allStreams.size()).toArray(new
DataStream[0]);
- commit(table, allStreams.get(0).union(rest));
+ commit(table, allStreams.get(0).union(rest),
CoreOptions.createCommitUser(userOptions));
return true;
}
@@ -322,12 +321,13 @@ public class SortedIndexTopoBuilder {
return new RowType(readType.isNullable(), fields);
}
- private static void commit(FileStoreTable table, DataStream<Committable>
written) {
+ private static void commit(
+ FileStoreTable table, DataStream<Committable> written, String
commitUser) {
OneInputStreamOperatorFactory<Committable, Committable>
committerOperator =
new CommitterOperatorFactory<>(
false,
true,
- "SortedIndexCommitter-" + UUID.randomUUID(),
+ commitUser,
context ->
new StoreCommitter(
table,
table.newCommit(context.commitUser()), context),
