From ecf3930a23b035c044b95c7c38dc23718314530e Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Sun, 30 Aug 2026 14:49:01 +0530 Subject: [PATCH 1/2] update --- .../paimon/iceberg/IcebergCommitCallback.java | 55 +++++++++++-------- .../iceberg/IcebergCompatibilityTest.java | 54 +++++++++++++++--- 2 files changed, 78 insertions(+), 31 deletions(-) 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 5b0a78b828fd..2b8f8fd0d6ac 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 @@ -70,6 +70,7 @@ import org.apache.paimon.types.MapType; import org.apache.paimon.types.MultisetType; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.BranchManager; import org.apache.paimon.utils.DataFilePathFactories; import org.apache.paimon.utils.FileStorePathFactory; import org.apache.paimon.utils.ManifestReadThreadPool; @@ -161,28 +162,31 @@ public IcebergCommitCallback(FileStoreTable table, String commitUser) { table.coreOptions().toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE); this.pathFactory = new IcebergPathFactory(catalogTableMetadataPath(table)); - IcebergMetadataCommitterFactory metadataCommitterFactory; - try { - metadataCommitterFactory = - FactoryUtil.discoverFactory( - IcebergCommitCallback.class.getClassLoader(), - IcebergMetadataCommitterFactory.class, - storageType.committerFactoryIdentifier()); - } catch (FactoryException e) { - metadataCommitterFactory = null; - // storage types without a committer have no factory by design, so a miss is expected - if (storageType.requiresMetadataCommitter()) { - LOG.warn( - "No IcebergMetadataCommitterFactory for '{}={}' found on the classpath, so " - + "table {} will not be synced to the external catalog (commits and " - + "metadata files are unaffected). Check that the module providing it " - + "is deployed and that its META-INF/services/{} entry survived " - + "shading. Cause: {}", - IcebergOptions.METADATA_ICEBERG_STORAGE.key(), - storageType, - table.fullName(), - Factory.class.getName(), - e.getMessage()); + IcebergMetadataCommitterFactory metadataCommitterFactory = null; + if (BranchManager.isMainBranch( + BranchManager.normalizeBranch(table.coreOptions().branch()))) { + try { + metadataCommitterFactory = + FactoryUtil.discoverFactory( + IcebergCommitCallback.class.getClassLoader(), + IcebergMetadataCommitterFactory.class, + storageType.committerFactoryIdentifier()); + } catch (FactoryException e) { + // storage types without a committer have no factory by design, so a miss is + // expected + if (storageType.requiresMetadataCommitter()) { + LOG.warn( + "No IcebergMetadataCommitterFactory for '{}={}' found on the classpath, so " + + "table {} will not be synced to the external catalog (commits and " + + "metadata files are unaffected). Check that the module providing it " + + "is deployed and that its META-INF/services/{} entry survived " + + "shading. Cause: {}", + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), + storageType, + table.fullName(), + Factory.class.getName(), + e.getMessage()); + } } } this.metadataCommitter = @@ -206,7 +210,12 @@ public IcebergCommitCallback(FileStoreTable table, String commitUser) { public static Path catalogTableMetadataPath(FileStoreTable table) { Path icebergDBPath = catalogDatabasePath(table); - return new Path(icebergDBPath, String.format("%s/metadata", table.location().getName())); + Path icebergTablePath = new Path(icebergDBPath, table.location().getName()); + return new Path( + BranchManager.branchPath( + icebergTablePath, + BranchManager.normalizeBranch(table.coreOptions().branch())), + "metadata"); } public static Path catalogDatabasePath(FileStoreTable table) { diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java index acaf154edcf5..5e558d6fcac3 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java @@ -1766,13 +1766,20 @@ public void testExistingTableWithUnpublishableHistoricalTimestampsRefusesToCommi } @Test - public void testCommitOnBranchMirrorsTheBranchSchema() throws Exception { + public void testCommitOnBranchUsesSeparateMetadata() 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); @@ -1785,7 +1792,22 @@ 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 mainMetadataPath = + new Path(IcebergCommitCallback.catalogTableMetadataPath(table), "v1.metadata.json"); + String mainMetadata = table.fileIO().readFileUtf8(mainMetadataPath); + 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)); @@ -1796,14 +1818,27 @@ public void testCommitOnBranchMirrorsTheBranchSchema() throws Exception { IcebergMetadata.fromPath( branchTable.fileIO(), new Path( - branchTable.location(), - "metadata/v" - + branchTable.snapshotManager().latestSnapshotId() - + ".metadata.json")); + IcebergCommitCallback.catalogTableMetadataPath(branchTable), + "v1.metadata.json")); assertThat( metadata.schemas().get(metadata.currentSchemaId()).fields().stream() .map(IcebergDataField::name)) .containsExactly("k", "v", "branch_only"); + assertThat(table.fileIO().readFileUtf8(mainMetadataPath)).isEqualTo(mainMetadata); + assertThat(table.fileIO().readFileUtf8(mainVersionHint)).isEqualTo(versionHint); + IcebergMetadata mainMetadataAfterBranchCommit = + IcebergMetadata.fromPath( + table.fileIO(), + new Path( + IcebergCommitCallback.catalogTableMetadataPath(table), + "v2.metadata.json")); + assertThat( + mainMetadataAfterBranchCommit.schemas() + .get(mainMetadataAfterBranchCommit.currentSchemaId()).fields() + .stream() + .map(IcebergDataField::name)) + .containsExactly("k", "v"); + assertThat(RecordingIcebergMetadataCommitter.COMMITS).isEmpty(); } /* @@ -2826,8 +2861,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); From 7a93cdc2b531a52c7490330410674298c105042e Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Sun, 30 Aug 2026 18:24:50 +0530 Subject: [PATCH 2/2] update --- .../org/apache/paimon/AbstractFileStore.java | 17 ++++-- .../paimon/iceberg/IcebergCommitCallback.java | 55 ++++++++----------- .../iceberg/IcebergCompatibilityTest.java | 37 +++---------- 3 files changed, 43 insertions(+), 66 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 b88798b9246d..ebf64c952068 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java @@ -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; @@ -398,8 +399,7 @@ private List 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; @@ -433,8 +433,7 @@ private List createCommitCallbacks(String commitUser, FileStoreT } } - if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE) - != IcebergOptions.StorageType.DISABLED) { + if (icebergCompatibilityEnabled(table)) { callbacks.add(new IcebergCommitCallback(table, commitUser)); } @@ -597,13 +596,19 @@ public List 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( + BranchManager.normalizeBranch(table.coreOptions().branch())); + } + @Override public ServiceManager newServiceManager() { return new ServiceManager(fileIO, options.path()); 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 2b8f8fd0d6ac..5b0a78b828fd 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 @@ -70,7 +70,6 @@ import org.apache.paimon.types.MapType; import org.apache.paimon.types.MultisetType; import org.apache.paimon.types.RowType; -import org.apache.paimon.utils.BranchManager; import org.apache.paimon.utils.DataFilePathFactories; import org.apache.paimon.utils.FileStorePathFactory; import org.apache.paimon.utils.ManifestReadThreadPool; @@ -162,31 +161,28 @@ public IcebergCommitCallback(FileStoreTable table, String commitUser) { table.coreOptions().toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE); this.pathFactory = new IcebergPathFactory(catalogTableMetadataPath(table)); - IcebergMetadataCommitterFactory metadataCommitterFactory = null; - if (BranchManager.isMainBranch( - BranchManager.normalizeBranch(table.coreOptions().branch()))) { - try { - metadataCommitterFactory = - FactoryUtil.discoverFactory( - IcebergCommitCallback.class.getClassLoader(), - IcebergMetadataCommitterFactory.class, - storageType.committerFactoryIdentifier()); - } catch (FactoryException e) { - // storage types without a committer have no factory by design, so a miss is - // expected - if (storageType.requiresMetadataCommitter()) { - LOG.warn( - "No IcebergMetadataCommitterFactory for '{}={}' found on the classpath, so " - + "table {} will not be synced to the external catalog (commits and " - + "metadata files are unaffected). Check that the module providing it " - + "is deployed and that its META-INF/services/{} entry survived " - + "shading. Cause: {}", - IcebergOptions.METADATA_ICEBERG_STORAGE.key(), - storageType, - table.fullName(), - Factory.class.getName(), - e.getMessage()); - } + IcebergMetadataCommitterFactory metadataCommitterFactory; + try { + metadataCommitterFactory = + FactoryUtil.discoverFactory( + IcebergCommitCallback.class.getClassLoader(), + IcebergMetadataCommitterFactory.class, + storageType.committerFactoryIdentifier()); + } catch (FactoryException e) { + metadataCommitterFactory = null; + // storage types without a committer have no factory by design, so a miss is expected + if (storageType.requiresMetadataCommitter()) { + LOG.warn( + "No IcebergMetadataCommitterFactory for '{}={}' found on the classpath, so " + + "table {} will not be synced to the external catalog (commits and " + + "metadata files are unaffected). Check that the module providing it " + + "is deployed and that its META-INF/services/{} entry survived " + + "shading. Cause: {}", + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), + storageType, + table.fullName(), + Factory.class.getName(), + e.getMessage()); } } this.metadataCommitter = @@ -210,12 +206,7 @@ public IcebergCommitCallback(FileStoreTable table, String commitUser) { public static Path catalogTableMetadataPath(FileStoreTable table) { Path icebergDBPath = catalogDatabasePath(table); - Path icebergTablePath = new Path(icebergDBPath, table.location().getName()); - return new Path( - BranchManager.branchPath( - icebergTablePath, - BranchManager.normalizeBranch(table.coreOptions().branch())), - "metadata"); + return new Path(icebergDBPath, String.format("%s/metadata", table.location().getName())); } public static Path catalogDatabasePath(FileStoreTable table) { diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java index 5e558d6fcac3..13d450c1db24 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java @@ -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; @@ -1766,7 +1765,7 @@ public void testExistingTableWithUnpublishableHistoricalTimestampsRefusesToCommi } @Test - public void testCommitOnBranchUsesSeparateMetadata() throws Exception { + public void testCommitAndTagOnBranchDoNotModifyMainMetadata() throws Exception { RecordingIcebergMetadataCommitter.COMMITS.clear(); RowType rowType = RowType.of( @@ -1798,9 +1797,11 @@ public void testCommitOnBranchUsesSeparateMetadata() throws Exception { commit.commit(2, write.prepareCommit(false, 2)); } - Path mainMetadataPath = - new Path(IcebergCommitCallback.catalogTableMetadataPath(table), "v1.metadata.json"); - String mainMetadata = table.fileIO().readFileUtf8(mainMetadataPath); + 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"); @@ -1813,31 +1814,11 @@ public void testCommitOnBranchUsesSeparateMetadata() throws Exception { 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( - IcebergCommitCallback.catalogTableMetadataPath(branchTable), - "v1.metadata.json")); - assertThat( - metadata.schemas().get(metadata.currentSchemaId()).fields().stream() - .map(IcebergDataField::name)) - .containsExactly("k", "v", "branch_only"); - assertThat(table.fileIO().readFileUtf8(mainMetadataPath)).isEqualTo(mainMetadata); + assertThat(table.fileIO().readFileUtf8(mainMetadataPathV1)).isEqualTo(mainMetadataV1); + assertThat(table.fileIO().readFileUtf8(mainMetadataPathV2)).isEqualTo(mainMetadataV2); assertThat(table.fileIO().readFileUtf8(mainVersionHint)).isEqualTo(versionHint); - IcebergMetadata mainMetadataAfterBranchCommit = - IcebergMetadata.fromPath( - table.fileIO(), - new Path( - IcebergCommitCallback.catalogTableMetadataPath(table), - "v2.metadata.json")); - assertThat( - mainMetadataAfterBranchCommit.schemas() - .get(mainMetadataAfterBranchCommit.currentSchemaId()).fields() - .stream() - .map(IcebergDataField::name)) - .containsExactly("k", "v"); assertThat(RecordingIcebergMetadataCommitter.COMMITS).isEmpty(); }