This is an automated email from the ASF dual-hosted git repository. yuqi1129 pushed a commit to branch review-11012 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 859634b68283ae1a18cd1d9387baf2ff8ff5c277 Author: yuqi <[email protected]> AuthorDate: Mon May 11 20:53:45 2026 +0800 Make Iceberg import idempotent under concurrency --- .../catalog/SchemaOperationDispatcher.java | 8 +++ .../catalog/TableOperationDispatcher.java | 13 +++++ .../catalog/TestSchemaOperationDispatcher.java | 31 ++++++++++++ .../catalog/TestTableOperationDispatcher.java | 58 ++++++++++++++++++++++ 4 files changed, 110 insertions(+) diff --git a/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java b/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java index b61c99e0d7..f034c40c3d 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java @@ -387,6 +387,14 @@ public class SchemaOperationDispatcher extends OperationDispatcher implements Sc try { store.put(schemaEntity, true); } catch (EntityAlreadyExistsException e) { + SchemaEntity concurrentSchemaEntity = getEntity(identifier, SCHEMA, SchemaEntity.class); + if (concurrentSchemaEntity != null) { + LOG.info( + "Schema {} was imported concurrently, reusing the existing entity in Gravitino.", + identifier); + return; + } + LOG.error("Failed to import schema {} with id {} to the store.", identifier, uid, e); throw new UnsupportedOperationException( "Schema managed by multiple catalogs. This may cause unexpected issues such as privilege conflicts. " diff --git a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java index 57499b6f6d..cb4aef4792 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java @@ -477,6 +477,19 @@ public class TableOperationDispatcher extends OperationDispatcher implements Tab try { store.put(tableEntity, true); } catch (EntityAlreadyExistsException e) { + TableEntity concurrentTableEntity = getEntity(identifier, TABLE, TableEntity.class); + if (concurrentTableEntity != null) { + LOG.info( + "Table {} was imported concurrently, reusing the existing entity in Gravitino.", + identifier); + return EntityCombinedTable.of(table.tableFromCatalog(), concurrentTableEntity) + .withHiddenProperties( + getHiddenPropertyNames( + getCatalogIdentifier(identifier), + HasPropertyMetadata::tablePropertiesMetadata, + table.tableFromCatalog().properties())); + } + LOG.error("Failed to import table {} with id {} to the store.", identifier, uid, e); throw new UnsupportedOperationException( "Table managed by multiple catalogs. This may cause unexpected issues such as privilege conflicts. " diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java b/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java index f57a8c4038..880450569c 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java @@ -40,6 +40,7 @@ import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.gravitino.Config; import org.apache.gravitino.Configs; import org.apache.gravitino.Entity; +import org.apache.gravitino.EntityAlreadyExistsException; import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; @@ -205,6 +206,36 @@ public class TestSchemaOperationDispatcher extends TestOperationDispatcher { Assertions.assertEquals("test", loadedSchema3.auditInfo().creator()); } + @Test + public void testConcurrentImportSchemaReusesExistingEntity() throws IOException { + NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog, "schemaConcurrent"); + Map<String, String> props = ImmutableMap.of("k1", "v1", "k2", "v2"); + dispatcher.createSchema(schemaIdent, "comment", props); + + AuditInfo concurrentAudit = + AuditInfo.builder().withCreator("concurrent").withCreateTime(Instant.now()).build(); + SchemaEntity concurrentSchemaEntity = + SchemaEntity.builder() + .withId(999L) + .withName(schemaIdent.name()) + .withNamespace(schemaIdent.namespace()) + .withAuditInfo(concurrentAudit) + .build(); + + reset(entityStore); + doThrow(new NoSuchEntityException("mock error")) + .doReturn(concurrentSchemaEntity) + .when(entityStore) + .get(any(), eq(Entity.EntityType.SCHEMA), any()); + doThrow(new EntityAlreadyExistsException("mock conflict")) + .when(entityStore) + .put(any(), anyBoolean()); + + Schema loadedSchema = Assertions.assertDoesNotThrow(() -> dispatcher.loadSchema(schemaIdent)); + Assertions.assertEquals(schemaIdent.name(), loadedSchema.name()); + Assertions.assertEquals("comment", loadedSchema.comment()); + } + @Test public void testCreateAndAlterSchema() throws IOException { NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog, "schema21"); diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java index d0016776c9..7e89a94747 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java @@ -44,9 +44,11 @@ import java.util.Map; import java.util.Optional; import java.util.function.Function; import java.util.stream.Collectors; +import java.util.stream.IntStream; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.gravitino.Config; import org.apache.gravitino.Entity; +import org.apache.gravitino.EntityAlreadyExistsException; import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; @@ -256,6 +258,62 @@ public class TestTableOperationDispatcher extends TestOperationDispatcher { Assertions.assertEquals("test", loadedTable4.auditInfo().creator()); } + @Test + public void testConcurrentImportTableReusesExistingEntity() throws IOException { + Namespace tableNs = Namespace.of(metalake, catalog, "schema52"); + Map<String, String> props = ImmutableMap.of("k1", "v1", "k2", "v2"); + schemaOperationDispatcher.createSchema(NameIdentifier.of(tableNs.levels()), "comment", props); + + NameIdentifier tableIdent = NameIdentifier.of(tableNs, "tableConcurrent"); + Column[] columns = + new Column[] { + TestColumn.builder() + .withName("col1") + .withPosition(0) + .withType(Types.StringType.get()) + .build(), + TestColumn.builder() + .withName("col2") + .withPosition(1) + .withType(Types.StringType.get()) + .build() + }; + + Table table = + tableOperationDispatcher.createTable( + tableIdent, columns, "comment", props, new Transform[0]); + + AuditInfo concurrentAudit = + AuditInfo.builder().withCreator("concurrent").withCreateTime(Instant.now()).build(); + TableEntity concurrentTableEntity = + TableEntity.builder() + .withId(999L) + .withName(tableIdent.name()) + .withNamespace(tableIdent.namespace()) + .withColumns( + IntStream.range(0, table.columns().length) + .mapToObj( + i -> + ColumnEntity.toColumnEntity(table.columns()[i], i, 0L, concurrentAudit)) + .collect(Collectors.toList())) + .withAuditInfo(concurrentAudit) + .build(); + + reset(entityStore); + doThrow(new NoSuchEntityException("mock error")) + .doReturn(concurrentTableEntity) + .when(entityStore) + .get(any(), eq(Entity.EntityType.TABLE), any()); + doThrow(new EntityAlreadyExistsException("mock conflict")) + .when(entityStore) + .put(any(), anyBoolean()); + + Table loadedTable = + Assertions.assertDoesNotThrow(() -> tableOperationDispatcher.loadTable(tableIdent)); + Assertions.assertEquals(tableIdent.name(), loadedTable.name()); + Assertions.assertEquals("comment", loadedTable.comment()); + } + @Test public void testCreateAndAlterTable() throws IOException { Namespace tableNs = Namespace.of(metalake, catalog, "schema61");
