From fc8dd84865c448dfdd998dec9e854cf90063506a Mon Sep 17 00:00:00 2001 From: Oleksandr Nitavskyi Date: Fri, 14 Aug 2026 17:09:54 +0200 Subject: [PATCH] [iceberg] Support custom table properties passthrough to Iceberg REST catalog Adds an explicit opt-in prefix, metadata.iceberg.table-properties., mirroring the existing metadata.iceberg.rest.* pattern. Custom properties are merged into the Iceberg table's properties on create, recreate (on metadata-base mismatch), and steady-state update paths, all of which route through IcebergRestMetadataCommitter#updateProperties(). Iceberg's reserved property keys are filtered out to avoid crashing the commit. ADPS-1261 --- .../apache/paimon/iceberg/IcebergOptions.java | 20 +++ .../iceberg/IcebergRestMetadataCommitter.java | 36 +++++ .../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 c28716dc25a8..819865066d12 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 METADATA_ICEBERG_STORAGE = key("metadata.iceberg.storage") .enumType(StorageType.class) @@ -188,6 +190,24 @@ public Map icebergRestConfig() { return restConfig; } + public Map icebergTableProperties() { + Map 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 2357a989c9ca..73cf9cd80057 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 @@ -64,6 +64,7 @@ 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 @@ -445,6 +446,12 @@ private void addAndSetCurrentSchema( // 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()); @@ -455,14 +462,43 @@ private void updateProperties(TableMetadata.Builder update) { || !desiredDeleteAfter.equals( current.get(METADATA_DELETE_AFTER_COMMIT_ENABLED)); + Map customProperties = customTableProperties(); + for (Map.Entry entry : customProperties.entrySet()) { + if (!entry.getValue().equals(current.get(entry.getKey()))) { + changed = true; + break; + } + } + if (changed) { Map 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., with + // Iceberg-reserved keys filtered out (Iceberg's TableMetadata rejects them outright). + private Map customTableProperties() { + Map customProperties = icebergOptions.icebergTableProperties(); + Map 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 052f76bda7ca..1375885587ab 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 @@ -494,6 +494,140 @@ public void testCommitAfterSchemaRollback() throws Exception { 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 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 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 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 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