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 d6ff29b7e8 [iceberg] Support custom table properties passthrough to
Iceberg REST catalog (#9268)
d6ff29b7e8 is described below
commit d6ff29b7e8cb67815656fd55bd0664252b0e9916
Author: Oleksandr Nitavskyi <[email protected]>
AuthorDate: Thu Aug 20 04:18:19 2026 +0200
[iceberg] Support custom table properties passthrough to Iceberg REST
catalog (#9268)
---
.../org/apache/paimon/iceberg/IcebergOptions.java | 20 +++
.../iceberg/IcebergRestMetadataCommitter.java | 36 ++++++
.../iceberg/IcebergRestMetadataCommitterTest.java | 134 +++++++++++++++++++++
3 files changed, 190 insertions(+)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
index c28716dc25..819865066d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
@@ -40,6 +40,8 @@ public class IcebergOptions {
public static final String REST_CONFIG_PREFIX = "metadata.iceberg.rest.";
+ public static final String TABLE_PROPERTIES_PREFIX =
"metadata.iceberg.table-properties.";
+
public static final ConfigOption<StorageType> METADATA_ICEBERG_STORAGE =
key("metadata.iceberg.storage")
.enumType(StorageType.class)
@@ -188,6 +190,24 @@ public class IcebergOptions {
return restConfig;
}
+ public Map<String, String> icebergTableProperties() {
+ Map<String, String> tableProperties = new HashMap<>();
+ options.keySet()
+ .forEach(
+ key -> {
+ if (key.startsWith(TABLE_PROPERTIES_PREFIX)) {
+ String propertyKey =
+
key.substring(TABLE_PROPERTIES_PREFIX.length());
+ Preconditions.checkArgument(
+ !propertyKey.isEmpty(),
+ "config key '%s' for iceberg table
property is empty!",
+ key);
+ tableProperties.put(propertyKey,
options.get(key));
+ }
+ });
+ return tableProperties;
+ }
+
public boolean deleteAfterCommitEnabled() {
return options.get(METADATA_DELETE_AFTER_COMMIT);
}
diff --git
a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
index 643839625c..bb86c53a6f 100644
---
a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
+++
b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
@@ -65,6 +65,7 @@ import java.util.stream.Collectors;
import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE;
import static
org.apache.iceberg.TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED;
import static
org.apache.iceberg.TableProperties.METADATA_PREVIOUS_VERSIONS_MAX;
+import static org.apache.iceberg.TableProperties.RESERVED_PROPERTIES;
/**
* commit Iceberg metadata to Iceberg's rest catalog, so the table can be
visited by Iceberg's rest
@@ -454,6 +455,12 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
// Update Iceberg REST table properties from current IcebergOptions, but
only
// if the values differ from what the REST catalog already has. This avoids
// emitting a redundant SetProperties update on every commit.
+ //
+ // This also merges in user-supplied custom properties
(metadata.iceberg.table-properties.*),
+ // since setProperties() merges into the existing property map rather than
replacing it
+ // (see TableMetadata.Builder#setProperties), so these persist across the
create, recreate,
+ // and steady-state update paths that all route through this method via
+ // updatesForCorrectBase().
private void updateProperties(TableMetadata.Builder update) {
String desiredMax =
String.valueOf(icebergOptions.previousVersionsMax());
String desiredDeleteAfter =
String.valueOf(icebergOptions.deleteAfterCommitEnabled());
@@ -464,14 +471,43 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
|| !desiredDeleteAfter.equals(
current.get(METADATA_DELETE_AFTER_COMMIT_ENABLED));
+ Map<String, String> customProperties = customTableProperties();
+ for (Map.Entry<String, String> entry : customProperties.entrySet()) {
+ if (!entry.getValue().equals(current.get(entry.getKey()))) {
+ changed = true;
+ break;
+ }
+ }
+
if (changed) {
Map<String, String> properties = new HashMap<>();
properties.put(METADATA_PREVIOUS_VERSIONS_MAX, desiredMax);
properties.put(METADATA_DELETE_AFTER_COMMIT_ENABLED,
desiredDeleteAfter);
+ properties.putAll(customProperties);
update.setProperties(properties);
}
}
+ // Custom table properties requested via
metadata.iceberg.table-properties.<key>, with
+ // Iceberg-reserved keys filtered out (Iceberg's TableMetadata rejects
them outright).
+ private Map<String, String> customTableProperties() {
+ Map<String, String> customProperties =
icebergOptions.icebergTableProperties();
+ Map<String, String> filtered = new HashMap<>();
+ customProperties.forEach(
+ (key, value) -> {
+ if (RESERVED_PROPERTIES.contains(key)) {
+ LOG.warn(
+ "Ignoring custom Iceberg table property '{}'
for table {}: "
+ + "it collides with an
Iceberg-reserved property.",
+ key,
+ icebergTableIdentifier);
+ } else {
+ filtered.put(key, value);
+ }
+ });
+ return filtered;
+ }
+
//
-------------------------------------------------------------------------------------
// Utils
//
-------------------------------------------------------------------------------------
diff --git
a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
index f374c8c119..15782acdc5 100644
---
a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
+++
b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
@@ -495,6 +495,140 @@ public class IcebergRestMetadataCommitterTest {
assertThat(icebergTable.currentSnapshot().schemaId()).isEqualTo(1);
}
+ @Test
+ public void testCustomTablePropertiesPassthrough() throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new
String[] {"k", "v"});
+ Map<String, String> customOptions = new HashMap<>();
+ customOptions.put(IcebergOptions.TABLE_PROPERTIES_PREFIX +
"dd.table-color", "blue");
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ 1,
+ randomFormat(),
+ customOptions);
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl<?> write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ // Custom property should be set on initial table creation.
+ write.write(GenericRow.of(1, 10));
+ commit.commit(1, write.prepareCommit(false, 1));
+ Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb",
"t"));
+ assertThat(icebergTable.properties()).containsEntry("dd.table-color",
"blue");
+
+ // Custom property should persist across a follow-up commit.
+ write.write(GenericRow.of(2, 20));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(2, write.prepareCommit(true, 2));
+ icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb", "t"));
+ assertThat(icebergTable.properties()).containsEntry("dd.table-color",
"blue");
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testCustomTablePropertyCollidingWithReservedKeyIsIgnored()
throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new
String[] {"k", "v"});
+ Map<String, String> customOptions = new HashMap<>();
+ // "format-version" is an Iceberg-reserved property key; Iceberg's
TableMetadata
+ // rejects it if present in a SetProperties update, so it must be
filtered out
+ // rather than crashing the commit.
+ customOptions.put(IcebergOptions.TABLE_PROPERTIES_PREFIX +
"format-version", "99");
+ customOptions.put(IcebergOptions.TABLE_PROPERTIES_PREFIX +
"dd.table-color", "green");
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ 1,
+ randomFormat(),
+ customOptions);
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl<?> write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 10));
+ commit.commit(1, write.prepareCommit(false, 1));
+
+ Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb",
"t"));
+ assertThat(icebergTable.properties()).containsEntry("dd.table-color",
"green");
+
assertThat(icebergTable.properties()).doesNotContainKey("format-version");
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testCustomTablePropertiesSurviveTableRecreate() throws
Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new
String[] {"k", "v"});
+ Map<String, String> customOptions = new HashMap<>();
+ customOptions.put(IcebergOptions.TABLE_PROPERTIES_PREFIX +
"dd.table-color", "blue");
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ 1,
+ randomFormat(),
+ customOptions);
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl<?> write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 10));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(1, write.prepareCommit(false, 1));
+
+ write.write(GenericRow.of(1, 11));
+ write.write(GenericRow.of(3, 30));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(2, write.prepareCommit(true, 2));
+
+ // Disable and re-enable Iceberg compatibility, forcing the REST
committer down the
+ // updatesForIncorrectBase() -> recreateTable() (drop-and-recreate)
path on the next
+ // commit, since the base metadata Paimon last wrote is now stale.
+ Map<String, String> options = new HashMap<>();
+ options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "disabled");
+ table = table.copy(options);
+ write.close();
+ write = table.newWrite(commitUser);
+ commit.close();
+ commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(4, 40));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(3, write.prepareCommit(true, 3));
+
+ options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
"rest-catalog");
+ table = table.copy(options);
+ write.close();
+ write = table.newWrite(commitUser);
+ commit.close();
+ commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(5, 50));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(4, write.prepareCommit(true, 4));
+
+ Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb",
"t"));
+ assertThat(icebergTable.properties()).containsEntry("dd.table-color",
"blue");
+
+ write.close();
+ commit.close();
+ }
+
@Test
public void testOptionOnlyAlterTableDoesNotCrashIcebergSync() throws
Exception {
// The fix deduplicates schemas in adjustMetadataForRest() and remaps
