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 75fc1819e3 [iceberg] Keep tables loadable when the mirror cannot
publish their schema (#10187)
75fc1819e3 is described below
commit 75fc1819e3fdc969f56b81d38a3f7144ec25be78
Author: Jiajia Li <[email protected]>
AuthorDate: Sat Oct 3 17:11:21 2026 +0800
[iceberg] Keep tables loadable when the mirror cannot publish their schema
(#10187)
---
.../paimon/iceberg/IcebergCommitCallback.java | 3 +
.../paimon/table/AbstractFileStoreTable.java | 12 +-
.../paimon/iceberg/IcebergCompatibilityTest.java | 126 +++++++++++++++++++++
3 files changed, 139 insertions(+), 2 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 ab73175396..08aaf110af 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
@@ -280,6 +280,9 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
@Override
public void retry(ManifestCommittable committable) {
+ // a retry runs no pre-commit validation and may republish metadata as
it stands
+ SchemaValidation.validateHistoricalIcebergTypes(
+ table.schemaManager()::listAll, table.coreOptions());
SnapshotManager snapshotManager = table.snapshotManager();
Snapshot snapshot =
snapshotManager
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index 93ecc173d5..5a04aeb116 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -397,7 +397,8 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
}
// validate schema with new options
- SchemaValidation.validateTableSchema(newTableSchema,
dynamicOptions.keySet());
+ SchemaValidation.validateTableSchema(
+ withoutIcebergMirror(newTableSchema), dynamicOptions.keySet());
if (new CoreOptions(tableSchema.options())
.toConfiguration()
.get(IcebergOptions.METADATA_ICEBERG_STORAGE)
@@ -423,13 +424,20 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
Map<String, String> mergedOptions = new
HashMap<>(latestSchema.options());
mergedOptions.putAll(tableSchema.options());
TableSchema newTableSchema = latestSchema.copy(mergedOptions);
- SchemaValidation.validateTableSchema(newTableSchema);
+
SchemaValidation.validateTableSchema(withoutIcebergMirror(newTableSchema));
return copy(newTableSchema);
} else {
return this;
}
}
+ /** Tables the mirror cannot publish must still load; every commit still
checks them. */
+ private static TableSchema withoutIcebergMirror(TableSchema schema) {
+ Map<String, String> options = new HashMap<>(schema.options());
+ options.remove(IcebergOptions.METADATA_ICEBERG_STORAGE.key());
+ return schema.copy(options);
+ }
+
@Override
public FileStoreTable copy(TableSchema newTableSchema) {
AbstractFileStoreTable copied =
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 8a2df8e51d..62e060489e 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
@@ -59,6 +59,7 @@ import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypeRoot;
@@ -113,6 +114,7 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for Iceberg compatibility. */
@@ -1718,6 +1720,130 @@ public class IcebergCompatibilityTest {
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(9));
}
+ @ParameterizedTest
+ @MethodSource("unpublishableTimestampTypes")
+ public void
testExistingTableWithUnpublishableTimestampsStillLoads(DataType timestampType)
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path warehouse = new Path(tempDir.toString());
+ Options options = new Options();
+ options.set(CoreOptions.BUCKET, 1);
+ options.set(CoreOptions.FILE_FORMAT, "parquet");
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), timestampType}, new
String[] {"k", "ts"});
+ Schema schema =
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ options.toMap(),
+ "");
+
+ Identifier identifier = Identifier.create("mydb", "t");
+ try (FileSystemCatalog paimonCatalog = new FileSystemCatalog(fileIO,
warehouse)) {
+ paimonCatalog.createDatabase("mydb", false);
+ paimonCatalog.createTable(identifier, schema, false);
+ FileStoreTable table = (FileStoreTable)
paimonCatalog.getTable(identifier);
+
+ String commitUser = UUID.randomUUID().toString();
+ try (TableWriteImpl<?> write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser)) {
+ write.write(GenericRow.of(1, Timestamp.fromEpochMillis(0)));
+ commit.commit(1, write.prepareCommit(false, 1));
+ }
+
+ Path tablePath = new Path(warehouse, "mydb.db/t");
+ TableSchema latest = new FileSystemSchemaManager(fileIO,
tablePath).latest().get();
+ Map<String, String> upgraded = new HashMap<>(latest.options());
+ upgraded.put(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.TABLE_LOCATION.toString());
+ TableSchema legacy =
+ new TableSchema(
+ latest.id() + 1,
+ latest.fields(),
+ latest.highestFieldId(),
+ latest.partitionKeys(),
+ latest.primaryKeys(),
+ upgraded,
+ latest.comment());
+ fileIO.writeFile(
+ new Path(tablePath, "schema/schema-" + legacy.id()),
legacy.toString(), true);
+
+ FileStoreTable loaded = (FileStoreTable)
paimonCatalog.getTable(identifier);
+ ReadBuilder readBuilder = loaded.newReadBuilder();
+ List<Integer> keys = new ArrayList<>();
+ readBuilder
+ .newRead()
+ .createReader(readBuilder.newScan().plan())
+ .forEachRemaining(row -> keys.add(row.getInt(0)));
+ assertThat(keys).containsExactly(1);
+ assertThatCode(() ->
loaded.copy(loaded.options())).doesNotThrowAnyException();
+
+ paimonCatalog.alterTable(
+ identifier,
+ SchemaChange.setOption(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.DISABLED.toString()),
+ false);
+ }
+ }
+
+ @Test
+ public void testRetryRefusesToRepublishWhatTheMirrorCannotPublish() 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.emptyList(), -1)
+ .copy(
+ Collections.singletonMap(
+
IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+
IcebergOptions.StorageType.HADOOP_CATALOG.toString()));
+
+ 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));
+ RecordingIcebergMetadataCommitter.failNextCommit = true;
+ write.write(GenericRow.of(2, 20));
+ assertThatThrownBy(() -> commit.commit(2, write.prepareCommit(false,
2)))
+ .hasStackTraceContaining("injected catalog failure");
+ write.close();
+ try {
+ commit.close();
+ } catch (Exception ignored) {
+ }
+
+ TableSchema latest = table.schemaManager().latest().get();
+ List<DataField> fields = new ArrayList<>(latest.fields());
+ fields.add(new DataField(latest.highestFieldId() + 1, "ts",
DataTypes.TIMESTAMP(9)));
+ TableSchema legacy =
+ new TableSchema(
+ latest.id() + 1,
+ fields,
+ latest.highestFieldId() + 1,
+ latest.partitionKeys(),
+ latest.primaryKeys(),
+ latest.options(),
+ latest.comment());
+ table.fileIO()
+ .writeFile(
+ new Path(table.location(), "schema/schema-" +
legacy.id()),
+ legacy.toString(),
+ true);
+
+ RecordingIcebergMetadataCommitter.COMMITS.clear();
+ IcebergCommitCallback callback = new IcebergCommitCallback(table,
commitUser);
+ assertThatThrownBy(() -> callback.retry(new ManifestCommittable(2)))
+ .hasMessageContaining("precision from 3 to 6");
+ callback.close();
+ assertThat(RecordingIcebergMetadataCommitter.COMMITS).isEmpty();
+ }
+
@ParameterizedTest
@MethodSource("unpublishableTimestampTypes")
public void
testExistingTableWithUnpublishableHistoricalTimestampsRefusesToCommit(
