Skip to content

[core] Prevent branch commits from overwriting Iceberg metadata #9476

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

Open
ArnavBalyan wants to merge 2 commits into
base: master
Choose a base branch
from
+47 −23
Open

[core] Prevent branch commits from overwriting Iceberg metadata #9476

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ecf3930
update
ArnavBalyan Aug 30, 2026
7a93cdc
update
ArnavBalyan Aug 30, 2026
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
17 changes: 11 additions & 6 deletions paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@
import org.apache.paimon.tag.TagAutoManager;
import org.apache.paimon.tag.TagPreview;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.BranchManager;
import org.apache.paimon.utils.ChainTableUtils;
import org.apache.paimon.utils.ChangelogManager;
import org.apache.paimon.utils.FileStorePathFactory;
Expand Down Expand Up @@ -398,8 +399,7 @@ private List<CommitPreCallback> createCommitPreCallbacks(FileStoreTable table) {
if (options.isChainTable()) {
callbacks.add(new ChainTableCommitPreCallback(table));
}
if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)
!= IcebergOptions.StorageType.DISABLED) {
if (icebergCompatibilityEnabled(table)) {
callbacks.add(new IcebergPreCommitValidation(table));
}
return callbacks;
Expand Down Expand Up @@ -433,8 +433,7 @@ private List<CommitCallback> createCommitCallbacks(String commitUser, FileStoreT
}
}

if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)
!= IcebergOptions.StorageType.DISABLED) {
if (icebergCompatibilityEnabled(table)) {
callbacks.add(new IcebergCommitCallback(table, commitUser));
}

Expand Down Expand Up @@ -597,13 +596,19 @@ public List<TagCallback> createTagCallbacks(FileStoreTable table) {
if (options.tagCreateSuccessFile()) {
callbacks.add(new SuccessFileTagCallback(fileIO, newTagManager().tagDirectory()));
}
if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)
!= IcebergOptions.StorageType.DISABLED) {
if (icebergCompatibilityEnabled(table)) {
callbacks.add(new IcebergCommitCallback(table, ""));
}
return callbacks;
}

private boolean icebergCompatibilityEnabled(FileStoreTable table) {
return options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)
!= IcebergOptions.StorageType.DISABLED
&& BranchManager.isMainBranch(

@JingsongLi JingsongLi Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Refresh Iceberg metadata when a branch is fast-forwarded to main

Disabling these callbacks on non-main branches prevents the premature overwrite, but it also leaves the normal promotion path without any Iceberg refresh. FileSystemBranchManager.fastForward only copies the branch snapshot/schema/tag files into the main directories and invalidates the Paimon snapshot cache; it never runs a main-branch IcebergCommitCallback. Therefore, after table.fastForward("b1"), Paimon main points at the promoted snapshot while the external Iceberg catalog and version-hint.text still point at the old main snapshot indefinitely. The new test stops before promotion, so it does not expose this stale-reader state.

Please make the fast-forward path rebuild/commit Iceberg metadata for the newly promoted main snapshot (while keeping branch commits callback-free), and add a regression that verifies the external pointer advances only after fast-forward.

BranchManager.normalizeBranch(table.coreOptions().branch()));
}

@Override
public ServiceManager newServiceManager() {
return new ServiceManager(fileIO, options.path());
Expand Down
53 changes: 36 additions & 17 deletions paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@
import org.apache.paimon.iceberg.manifest.IcebergManifestFile;
import org.apache.paimon.iceberg.manifest.IcebergManifestFileMeta;
import org.apache.paimon.iceberg.manifest.IcebergManifestList;
import org.apache.paimon.iceberg.metadata.IcebergDataField;
import org.apache.paimon.iceberg.metadata.IcebergMetadata;
import org.apache.paimon.iceberg.metadata.IcebergRef;
import org.apache.paimon.iceberg.metadata.IcebergSchema;
Expand Down Expand Up @@ -1766,13 +1765,20 @@ public void testExistingTableWithUnpublishableHistoricalTimestampsRefusesToCommi
}

@Test
public void testCommitOnBranchMirrorsTheBranchSchema() throws Exception {
public void testCommitAndTagOnBranchDoNotModifyMainMetadata() throws Exception {
RecordingIcebergMetadataCommitter.COMMITS.clear();
RowType rowType =
RowType.of(
new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
FileStoreTable table =
createPaimonTable(
rowType, Collections.emptyList(), Collections.singletonList("k"), 1);
rowType,
Collections.emptyList(),
Collections.singletonList("k"),
1,
Collections.singletonMap(
IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
IcebergOptions.StorageType.HADOOP_CATALOG.toString()));

String commitUser = UUID.randomUUID().toString();
try (TableWriteImpl<?> write = table.newWrite(commitUser);
Expand All @@ -1785,25 +1791,35 @@ public void testCommitOnBranchMirrorsTheBranchSchema() throws Exception {
new SchemaManager(table.fileIO(), table.location(), "b1")
.commitChanges(SchemaChange.addColumn("branch_only", DataTypes.INT()));

try (TableWriteImpl<?> write = table.newWrite(commitUser);
TableCommitImpl commit = table.newCommit(commitUser)) {
write.write(GenericRow.of(3, 30));
commit.commit(2, write.prepareCommit(false, 2));
}

Path metadataDirectory = IcebergCommitCallback.catalogTableMetadataPath(table);
Path mainMetadataPathV1 = new Path(metadataDirectory, "v1.metadata.json");
String mainMetadataV1 = table.fileIO().readFileUtf8(mainMetadataPathV1);
Path mainMetadataPathV2 = new Path(metadataDirectory, "v2.metadata.json");
String mainMetadataV2 = table.fileIO().readFileUtf8(mainMetadataPathV2);
Path mainVersionHint =
new Path(
IcebergCommitCallback.catalogTableMetadataPath(table), "version-hint.text");
String versionHint = table.fileIO().readFileUtf8(mainVersionHint);

FileStoreTable branchTable = table.switchToBranch("b1");
RecordingIcebergMetadataCommitter.COMMITS.clear();
try (TableWriteImpl<?> write = branchTable.newWrite(commitUser);
TableCommitImpl commit = branchTable.newCommit(commitUser)) {
write.write(GenericRow.of(2, 20, 200));
commit.commit(2, write.prepareCommit(false, 2));
}
branchTable.createTag("branch-tag");

IcebergMetadata metadata =
IcebergMetadata.fromPath(
branchTable.fileIO(),
new Path(
branchTable.location(),
"metadata/v"
+ branchTable.snapshotManager().latestSnapshotId()
+ ".metadata.json"));
assertThat(
metadata.schemas().get(metadata.currentSchemaId()).fields().stream()
.map(IcebergDataField::name))
.containsExactly("k", "v", "branch_only");
assertThat(table.fileIO().readFileUtf8(mainMetadataPathV1)).isEqualTo(mainMetadataV1);
assertThat(table.fileIO().readFileUtf8(mainMetadataPathV2)).isEqualTo(mainMetadataV2);
assertThat(table.fileIO().readFileUtf8(mainVersionHint)).isEqualTo(versionHint);
assertThat(RecordingIcebergMetadataCommitter.COMMITS).isEmpty();
}

/*
Expand Down Expand Up @@ -2826,8 +2842,11 @@ private FileStoreTable createPaimonTable(

Options options = new Options(customOptions);
options.set(CoreOptions.BUCKET, numBuckets);
options.set(
IcebergOptions.METADATA_ICEBERG_STORAGE, IcebergOptions.StorageType.TABLE_LOCATION);
if (!options.contains(IcebergOptions.METADATA_ICEBERG_STORAGE)) {
options.set(
IcebergOptions.METADATA_ICEBERG_STORAGE,
IcebergOptions.StorageType.TABLE_LOCATION);
}
options.set(CoreOptions.FILE_FORMAT, "avro");
options.set(CoreOptions.TARGET_FILE_SIZE, MemorySize.ofKibiBytes(32));
options.set(IcebergOptions.COMPACT_MIN_FILE_NUM, 4);
Expand Down
Loading