Skip to content

[iceberg] Support custom table properties passthrough to Iceberg REST catalog #9268

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Aug 20, 2026
Merged

[iceberg] Support custom table properties passthrough to Iceberg REST catalog #9268

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -188,6 +190,24 @@ public Map<String, String> icebergRestConfig() {
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);
}
Expand Down
36 changes: 36 additions & 0 deletions paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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());
Expand All @@ -455,14 +462,43 @@ private void updateProperties(TableMetadata.Builder update) {
|| !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
// -------------------------------------------------------------------------------------
Expand Down
134 changes: 134 additions & 0 deletions paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<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
Expand Down
Loading