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 f63c1d8b39 [core] Custom tag callback support to be initialized with
table (#8189)
f63c1d8b39 is described below
commit f63c1d8b39eaa8b13e5b3b17b0b0b1f77e466de0
Author: yuzelin <[email protected]>
AuthorDate: Wed Jun 10 15:11:01 2026 +0800
[core] Custom tag callback support to be initialized with table (#8189)
---
.../src/main/java/org/apache/paimon/AbstractFileStore.java | 3 ++-
.../java/org/apache/paimon/iceberg/IcebergCommitCallback.java | 5 +++++
.../src/main/java/org/apache/paimon/table/sink/CallbackUtils.java | 8 ++++++--
.../src/main/java/org/apache/paimon/table/sink/TagCallback.java | 4 ++++
4 files changed, 17 insertions(+), 3 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
index 3aa399186e..cc298823ff 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -566,7 +566,8 @@ abstract class AbstractFileStore<T> implements FileStore<T>
{
@Override
public List<TagCallback> createTagCallbacks(FileStoreTable table) {
- List<TagCallback> callbacks = new
ArrayList<>(CallbackUtils.loadTagCallbacks(options));
+ List<TagCallback> callbacks =
+ new ArrayList<>(CallbackUtils.loadTagCallbacks(options,
table));
String partitionField = options.tagToPartitionField();
if (partitionField != null) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
index 0489a2d0bf..79c2ecb7ad 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
@@ -271,6 +271,11 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
indexFileHandler.scan(snapshot, DELETION_VECTORS_INDEX));
}
+ @Override
+ public void setTable(FileStoreTable table) {
+ // nothing to do
+ }
+
private void createMetadata(
Snapshot snapshot,
FileChangesCollector fileChangesCollector,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/CallbackUtils.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/CallbackUtils.java
index bec038718c..de27922e86 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/CallbackUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/CallbackUtils.java
@@ -29,8 +29,12 @@ import java.util.Map;
/** Utils to load callbacks. */
public class CallbackUtils {
- public static List<TagCallback> loadTagCallbacks(CoreOptions coreOptions) {
- return loadCallbacks(coreOptions.tagCallbacks(), TagCallback.class);
+ public static List<TagCallback> loadTagCallbacks(
+ CoreOptions coreOptions, FileStoreTable table) {
+ List<TagCallback> tagCallbacks =
+ loadCallbacks(coreOptions.tagCallbacks(), TagCallback.class);
+ tagCallbacks.forEach(callback -> callback.setTable(table));
+ return tagCallbacks;
}
public static List<CommitCallback> loadCommitCallbacks(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TagCallback.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TagCallback.java
index d99451633e..a0f2643f0e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TagCallback.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TagCallback.java
@@ -18,6 +18,8 @@
package org.apache.paimon.table.sink;
+import org.apache.paimon.table.FileStoreTable;
+
/**
* This callback will be called after tag operations.
*
@@ -25,6 +27,8 @@ package org.apache.paimon.table.sink;
*/
public interface TagCallback extends AutoCloseable {
+ default void setTable(FileStoreTable table) {}
+
void notifyCreation(String tagName);
// Iceberg tag callbacks require snapshotId
