This is an automated email from the ASF dual-hosted git repository. yuqi1129 pushed a commit to branch feat/12345-occ-table in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 23d0f3c15f07b52c25817f51cf46807db1e8a1c8 Author: yuqi <[email protected]> AuthorDate: Fri Aug 21 17:28:31 2026 +0800 [#12345] improvement(core): add OCC for table writes --- .../lakehouse/lance/LanceTableOperations.java | 40 +- .../lance/TestLanceConcurrentRepairStress.java | 19 +- .../lakehouse/lance/TestLanceTableOperations.java | 94 +++- .../gravitino/catalog/OperationDispatcher.java | 4 + .../storage/relational/mapper/TableMetaMapper.java | 19 +- .../mapper/TableMetaSQLProviderFactory.java | 22 +- .../provider/base/TableMetaBaseSQLProvider.java | 50 +- .../postgresql/TableMetaPostgreSQLProvider.java | 5 +- .../relational/service/TableMetaService.java | 208 +++++---- .../catalog/TestTableOperationDispatcher.java | 12 +- .../base/TestTableMetaBaseSQLProvider.java | 55 +++ .../TestTableMetaPostgreSQLProvider.java | 33 ++ .../relational/service/TestTableMetaService.java | 511 +++++++++++++++++++++ 13 files changed, 933 insertions(+), 139 deletions(-) diff --git a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java index 7edfc468d5..08b36cae5e 100644 --- a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java +++ b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java @@ -50,6 +50,7 @@ import org.apache.gravitino.connector.SupportsSchemas; import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.exceptions.NoSuchSchemaException; import org.apache.gravitino.exceptions.NoSuchTableException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.exceptions.TableAlreadyExistsException; import org.apache.gravitino.lance.common.ops.gravitino.LanceDataTypeConverter; import org.apache.gravitino.lance.common.utils.LanceConstants; @@ -65,7 +66,6 @@ import org.apache.gravitino.rel.expressions.sorts.SortOrder; import org.apache.gravitino.rel.expressions.transforms.Transform; import org.apache.gravitino.rel.indexes.Index; import org.apache.gravitino.storage.IdGenerator; -import org.apache.gravitino.storage.relational.service.TableMetaService; import org.apache.gravitino.utils.PrincipalUtils; import org.lance.Dataset; import org.lance.ReadOptions; @@ -561,30 +561,20 @@ public class LanceTableOperations extends ManagedTableOperations { } /** - * Applies an idempotent update to the stored table, retrying when the optimistic-lock CAS is lost - * to a concurrent update. The repair-on-load path runs on every {@code loadTable}, so concurrent - * loads of the same table race on the version CAS; {@code store.update} surfaces the lost race as - * an {@link IOException} whose message starts with {@link - * TableMetaService#UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX}. Because the updater is idempotent, the - * loser sleeps a short randomized backoff (to avoid re-colliding), re-reads the latest (already - * repaired) entity, and retries instead of failing the whole load with a fatal error. Other IO - * failures (DB outage, serialization errors, etc.) are not conflicts and fail fast. + * Repairs stored table metadata and retries only when another repair wins the same race. + * + * <p>Two concurrent loads can both read the old metadata. The first update wins; the second gets + * {@link OptimisticLockException}. Repair is safe to run more than once, so the loser waits + * briefly, reads the winner's latest row, and tries again. Ordinary IO failures are not safe to + * retry here and are returned immediately. */ private TableEntity updateTableWithCasRetry( NameIdentifier ident, Function<TableEntity, TableEntity> updater) throws IOException { - IOException lastConflict = null; + OptimisticLockException lastConflict = null; for (int attempt = 1; attempt <= REPAIR_UPDATE_MAX_ATTEMPTS; attempt++) { try { return store.update(ident, TableEntity.class, Entity.EntityType.TABLE, updater); - } catch (IOException e) { - // Only retry when the update matched 0 rows (lost optimistic-lock CAS). Other IO failures - // (DB outage, serialization errors, etc.) should fail fast. - String message = e.getMessage(); - if (message == null - || !message.startsWith(TableMetaService.UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX)) { - throw e; - } - + } catch (OptimisticLockException e) { lastConflict = e; LOG.debug( "Optimistic-lock conflict updating table {} metadata (attempt {}/{}), {}", @@ -599,11 +589,13 @@ public class LanceTableOperations extends ManagedTableOperations { } } } - throw new IOException( - String.format( - "Failed to update table %s after %d optimistic-lock retries", - ident, REPAIR_UPDATE_MAX_ATTEMPTS), - lastConflict); + // Reaching here means every attempt lost to another writer. Keep the exception type so the + // caller still knows this is an OCC conflict, and add the attempt count for diagnosis. + throw new OptimisticLockException( + lastConflict, + "Failed to repair table %s after %d optimistic-lock attempts", + ident, + REPAIR_UPDATE_MAX_ATTEMPTS); } /** diff --git a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceConcurrentRepairStress.java b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceConcurrentRepairStress.java index 4a67be530b..1921d31d3f 100644 --- a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceConcurrentRepairStress.java +++ b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceConcurrentRepairStress.java @@ -50,6 +50,7 @@ import org.apache.gravitino.HasIdentifier; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.UserPrincipal; import org.apache.gravitino.catalog.ManagedSchemaOperations; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.TableEntity; import org.apache.gravitino.rel.Table; @@ -66,16 +67,16 @@ import org.mockito.Mockito; /** * Real multi-threaded reproduction of the repair-on-load optimistic-lock race behind #11891. Unlike * {@code TestLanceTableOperations#testLoadTableSurvivesConcurrentRepairVersionRace} (a - * deterministic mock that throws one scripted {@code IOException}), this test drives {@link + * deterministic mock that throws one scripted conflict), this test drives {@link * LanceTableOperations#loadTable} from several threads at once against a {@link CasEntityStore} * that models the production relational store's compare-and-set semantics faithfully: every {@code * update} bumps a version guarded by the base version, so concurrent updates from the same base * conflict and exactly one wins per generation — just like {@code TableMetaService.updateTable} * ({@code UPDATE ... WHERE current_version = old}). * - * <p>Before the CAS retry, the loser of the race got {@code IOException("Failed to update the - * entity")}, rethrown as a fatal {@code RuntimeException} (HTTP 500). This test asserts every - * concurrent load returns the repaired table instead. + * <p>Before the CAS retry, the loser of the race got an {@link OptimisticLockException}, rethrown + * as a fatal error (HTTP 500). This test asserts every concurrent load returns the repaired table + * instead. */ public class TestLanceConcurrentRepairStress { @@ -189,10 +190,10 @@ public class TestLanceConcurrentRepairStress { /** * In-memory {@link EntityStore} that reproduces the relational store's optimistic-lock CAS: * {@code update} reads a versioned snapshot, applies the (idempotent) updater, and commits only - * if the version has not advanced since the read — otherwise it throws {@code IOException("Failed - * to update the entity")}, exactly as {@code TableMetaService.updateTable} does when {@code - * UPDATE ... WHERE current_version = old} matches zero rows. Every commit bumps the version, so - * even a no-op update invalidates a concurrent update from the same base, matching production. + * if the version has not advanced since the read — otherwise it throws an {@link + * OptimisticLockException}, exactly as the relational table service does when {@code UPDATE ... + * WHERE current_version = old} matches zero rows. Every commit bumps the version, so even a no-op + * update invalidates a concurrent update from the same base, matching production. */ private static final class CasEntityStore implements EntityStore { @@ -230,7 +231,7 @@ public class TestLanceConcurrentRepairStress { if (ref.compareAndSet(base, next)) { return updated; } - throw new IOException("Failed to update the entity: " + ident); + throw new OptimisticLockException("mock conflict for %s", ident); } // --- unused surface --------------------------------------------------- diff --git a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java index bf99a64814..e0226c1b4d 100644 --- a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java +++ b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java @@ -48,6 +48,7 @@ import org.apache.gravitino.EntityStore; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.UserPrincipal; import org.apache.gravitino.catalog.ManagedSchemaOperations; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.ColumnEntity; import org.apache.gravitino.meta.TableEntity; @@ -174,10 +175,10 @@ public class TestLanceTableOperations { /** * Reproduces the concurrent repair-on-load race seen in {@code LanceSparkRESTServiceIT}. When two * loads repair the same table at once, the optimistic-locked {@code store.update} of the slower - * one matches zero rows and {@code TableMetaService} surfaces it as {@code IOException("Failed to - * update the entity")}. Before the CAS retry, {@code repairTableMetadata} rethrew it as a fatal - * {@code RuntimeException} (HTTP 500) instead of tolerating the concurrent update. This test - * asserts that the lost race is benign and load returns a usable table. + * one matches zero rows and the store surfaces an {@link OptimisticLockException}. Before the CAS + * retry, {@code repairTableMetadata} rethrew it as a fatal error (HTTP 500) instead of tolerating + * the concurrent update. This test asserts that the lost race is benign and load returns a usable + * table. */ @Test public void testLoadTableSurvivesConcurrentRepairVersionRace() throws Exception { @@ -219,10 +220,10 @@ public class TestLanceTableOperations { when(idGenerator.nextId()).thenReturn(10L, 11L); // First repair attempt loses the optimistic-lock CAS (a concurrent load already bumped the - // version): TableMetaService surfaces exactly this IOException. The retry re-reads the winner's - // already-repaired entity, against which the idempotent updater succeeds. + // version). The retry re-reads the winner's already-repaired entity, against which the + // idempotent updater succeeds. when(store.update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE), any())) - .thenThrow(new IOException("Failed to update the entity: " + ident)) + .thenThrow(new OptimisticLockException("mock conflict")) .thenAnswer( invocation -> { @SuppressWarnings("unchecked") @@ -255,6 +256,58 @@ public class TestLanceTableOperations { Assertions.assertEquals("name", loadedTable.columns()[1].name()); } + @Test + public void testRepairStopsAfterBoundedOptimisticLockRetries() throws Exception { + NameIdentifier ident = prepareDeclaredTableForRepair("repair-conflict-exhausted"); + when(store.update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE), any())) + .thenThrow(new OptimisticLockException("mock conflict")); + // Remove the real sleep so this test checks the retry bound without becoming timing-sensitive. + Mockito.doNothing().when(lanceTableOps).backoffBeforeRetry(any()); + + OptimisticLockException failure = + Assertions.assertThrows( + OptimisticLockException.class, + () -> + PrincipalUtils.doAs( + new UserPrincipal("tester"), () -> lanceTableOps.loadTable(ident))); + Assertions.assertInstanceOf(OptimisticLockException.class, failure.getCause()); + verify(store, Mockito.times(5)) + .update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE), any()); + } + + @Test + public void testRepairDoesNotRetryOrdinaryIoFailure() throws Exception { + NameIdentifier ident = prepareDeclaredTableForRepair("repair-io-failure"); + when(store.update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE), any())) + .thenThrow(new IOException("database unavailable")); + + RuntimeException failure = + Assertions.assertThrows( + RuntimeException.class, + () -> + PrincipalUtils.doAs( + new UserPrincipal("tester"), () -> lanceTableOps.loadTable(ident))); + Assertions.assertInstanceOf(IOException.class, failure.getCause()); + verify(store, Mockito.times(1)) + .update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE), any()); + } + + @Test + public void testRepairBackoffPreservesThreadInterrupt() { + NameIdentifier ident = NameIdentifier.of("schema", "table"); + Thread.currentThread().interrupt(); + try { + IOException failure = + Assertions.assertThrows(IOException.class, () -> lanceTableOps.backoffBeforeRetry(ident)); + + Assertions.assertInstanceOf(InterruptedException.class, failure.getCause()); + Assertions.assertTrue(Thread.currentThread().isInterrupted()); + } finally { + // JUnit reuses worker threads, so do not leak this test's interrupt flag into another test. + Thread.interrupted(); + } + } + @Test public void testLoadTableWithStoredColumnsDoesNotReadLocation() throws Exception { NameIdentifier ident = NameIdentifier.of("schema", "table"); @@ -954,4 +1007,31 @@ public class TestLanceTableOperations { AuditInfo.builder().withCreator("creator").withCreateTime(Instant.EPOCH).build()) .build(); } + + private NameIdentifier prepareDeclaredTableForRepair(String directoryName) throws Exception { + NameIdentifier ident = NameIdentifier.of("schema", "table"); + String location = tempDir.resolve(directoryName).toString(); + TableEntity tableEntity = + tableEntity( + ident, + List.of(), + Map.of( + Table.PROPERTY_LOCATION, + location, + LANCE_TABLE_DECLARED, + "true", + LANCE_STORAGE_OPTIONS_PREFIX + "endpoint", + "http://endpoint";)); + when(store.get(eq(ident), eq(Entity.EntityType.TABLE), eq(TableEntity.class))) + .thenReturn(tableEntity); + + Dataset dataset = mock(Dataset.class); + when(dataset.getSchema()) + .thenReturn(new Schema(List.of(Field.nullable("id", new ArrowType.Int(32, true))))); + when(dataset.version()).thenReturn(8L); + Mockito.doReturn(dataset) + .when(lanceTableOps) + .openDataset(location, Map.of("endpoint", "http://endpoint";)); + return ident; + } } diff --git a/core/src/main/java/org/apache/gravitino/catalog/OperationDispatcher.java b/core/src/main/java/org/apache/gravitino/catalog/OperationDispatcher.java index 4fa1808fc6..adf6d64482 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/OperationDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/catalog/OperationDispatcher.java @@ -35,6 +35,7 @@ import org.apache.gravitino.connector.HasPropertyMetadata; import org.apache.gravitino.connector.PropertiesMetadata; import org.apache.gravitino.connector.capability.Capability; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.file.FilesetChange; import org.apache.gravitino.messaging.TopicChange; import org.apache.gravitino.rel.SupportsPartitions; @@ -219,6 +220,9 @@ public abstract class OperationDispatcher { R ret = null; try { ret = fn.apply(ident); + } catch (OptimisticLockException e) { + // A version conflict is actionable: the caller must retry against the latest entity. + throw e; } catch (NoSuchEntityException e) { // Case 2: The table is created by Gravitino, but has no corresponding entity in Gravitino. LOG.error(FormattedErrorMessages.ENTITY_NOT_FOUND, ident); diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.java index acb682916d..4fdc5103ae 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.java @@ -73,6 +73,15 @@ public interface TableMetaMapper { @Param("schemaName") String schemaName, @Param("tableName") String tableName); + /** + * Selects and exclusively locks an active table metadata row. + * + * @param tableId the table ID + * @return the active table metadata, or {@code null} when it no longer exists + */ + @SelectProvider(type = TableMetaSQLProviderFactory.class, method = "selectTableMetaByIdForUpdate") + TablePO selectTableMetaByIdForUpdate(@Param("tableId") Long tableId); + @InsertProvider(type = TableMetaSQLProviderFactory.class, method = "insertTableMeta") void insertTableMeta(@Param("tableMeta") TablePO tablePO); @@ -87,10 +96,18 @@ public interface TableMetaMapper { @Param("oldTableMeta") TablePO oldTablePO, @Param("newSchemaId") Long newSchemaId); + /** + * Soft-deletes a table only if its version has not changed since the caller read it. + * + * @param tableId the table ID + * @param currentVersion the version observed by the caller + * @return the number of deleted rows; zero means the table changed or disappeared + */ @UpdateProvider( type = TableMetaSQLProviderFactory.class, method = "softDeleteTableMetasByTableId") - Integer softDeleteTableMetasByTableId(@Param("tableId") Long tableId); + Integer softDeleteTableMetasByTableId( + @Param("tableId") Long tableId, @Param("currentVersion") Long currentVersion); @UpdateProvider( type = TableMetaSQLProviderFactory.class, diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaSQLProviderFactory.java index c69fd9a191..2876bbc1d1 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaSQLProviderFactory.java @@ -89,6 +89,16 @@ public class TableMetaSQLProviderFactory { return getProvider().selectTableMetaById(tableId); } + /** + * Returns SQL that selects and exclusively locks an active table metadata row. + * + * @param tableId the table ID + * @return the locking select SQL + */ + public static String selectTableMetaByIdForUpdate(@Param("tableId") Long tableId) { + return getProvider().selectTableMetaByIdForUpdate(tableId); + } + public static String insertTableMeta(@Param("tableMeta") TablePO tablePO) { return getProvider().insertTableMeta(tablePO); } @@ -104,8 +114,16 @@ public class TableMetaSQLProviderFactory { return getProvider().updateTableMeta(newTablePO, oldTablePO, newSchemaId); } - public static String softDeleteTableMetasByTableId(@Param("tableId") Long tableId) { - return getProvider().softDeleteTableMetasByTableId(tableId); + /** + * Returns SQL that soft-deletes a table with a version check. + * + * @param tableId the table ID + * @param currentVersion the version observed by the caller + * @return the version-checked delete SQL + */ + public static String softDeleteTableMetasByTableId( + @Param("tableId") Long tableId, @Param("currentVersion") Long currentVersion) { + return getProvider().softDeleteTableMetasByTableId(tableId, currentVersion); } public static String softDeleteTableMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TableMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TableMetaBaseSQLProvider.java index 06684724fa..71f63cb2b6 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TableMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TableMetaBaseSQLProvider.java @@ -178,6 +178,23 @@ public class TableMetaBaseSQLProvider { + " WHERE tm.table_id = #{tableId} AND tm.deleted_at = 0"; } + /** + * Returns the active table metadata row and holds it exclusively for the current transaction. + * + * @param tableId the table ID + * @return the locking select SQL + */ + public String selectTableMetaByIdForUpdate(@Param("tableId") Long tableId) { + return "SELECT table_id as tableId, table_name as tableName," + + " metalake_id as metalakeId, catalog_id as catalogId," + + " schema_id as schemaId, audit_info as auditInfo," + + " current_version as currentVersion, last_version as lastVersion," + + " deleted_at as deletedAt" + + " FROM " + + TABLE_NAME + + " WHERE table_id = #{tableId} AND deleted_at = 0 FOR UPDATE"; + } + public String insertTableMeta(@Param("tableMeta") TablePO tablePO) { return "INSERT INTO " + TABLE_NAME @@ -225,6 +242,18 @@ public class TableMetaBaseSQLProvider { + " deleted_at = #{tableMeta.deletedAt}"; } + /** + * Returns SQL that updates a table only while its OCC version is unchanged. + * + * <p>The WHERE clause intentionally compares only the stable ID, the version seen by the caller, + * and the active flag. Comparing serialized payload or audit fields would make harmless encoding + * differences look like conflicts; the version is the single source of truth for concurrency. + * + * @param newTablePO the table values to write + * @param oldTablePO the table values and OCC version read by the caller + * @param newSchemaId the target schema ID + * @return the version-checked update SQL + */ public String updateTableMeta( @Param("newTableMeta") TablePO newTablePO, @Param("oldTableMeta") TablePO oldTablePO, @@ -240,22 +269,27 @@ public class TableMetaBaseSQLProvider { + " last_version = #{newTableMeta.lastVersion}," + " deleted_at = #{newTableMeta.deletedAt}" + " WHERE table_id = #{oldTableMeta.tableId}" - + " AND table_name = #{oldTableMeta.tableName}" - + " AND metalake_id = #{oldTableMeta.metalakeId}" - + " AND catalog_id = #{oldTableMeta.catalogId}" - + " AND schema_id = #{oldTableMeta.schemaId}" - + " AND audit_info = #{oldTableMeta.auditInfo}" + " AND current_version = #{oldTableMeta.currentVersion}" - + " AND last_version = #{oldTableMeta.lastVersion}" + " AND deleted_at = 0"; } - public String softDeleteTableMetasByTableId(@Param("tableId") Long tableId) { + /** + * Returns SQL that deletes only the table version observed by the caller. + * + * <p>For example, a drop that read version 3 must not delete version 4 after a concurrent alter. + * + * @param tableId the table ID + * @param currentVersion the version observed by the caller + * @return the version-checked delete SQL + */ + public String softDeleteTableMetasByTableId( + @Param("tableId") Long tableId, @Param("currentVersion") Long currentVersion) { return "UPDATE " + TABLE_NAME + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)" + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000" - + " WHERE table_id = #{tableId} AND deleted_at = 0"; + + " WHERE table_id = #{tableId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; } public String softDeleteTableMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TableMetaPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TableMetaPostgreSQLProvider.java index 7add18edac..13b16579f9 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TableMetaPostgreSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TableMetaPostgreSQLProvider.java @@ -56,11 +56,12 @@ public class TableMetaPostgreSQLProvider extends TableMetaBaseSQLProvider { } @Override - public String softDeleteTableMetasByTableId(Long tableId) { + public String softDeleteTableMetasByTableId(Long tableId, Long currentVersion) { return "UPDATE " + TABLE_NAME + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)" - + " WHERE table_id = #{tableId} AND deleted_at = 0"; + + " WHERE table_id = #{tableId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; } @Override diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java index 741a210e10..a2e0fdf25e 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java @@ -25,8 +25,6 @@ import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.Objects; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import org.apache.gravitino.Entity; import org.apache.gravitino.HasIdentifier; @@ -55,15 +53,6 @@ import org.apache.gravitino.utils.NamespaceUtil; /** The service class for table metadata. It provides the basic database operations for table. */ public class TableMetaService { - /** - * Message prefix of the {@link java.io.IOException} thrown by {@link #updateTable} when the - * optimistic-lock CAS matches zero rows (the stored version advanced under a concurrent update). - * Exposed so callers that retry the lost race (e.g. the Lance repair-on-load path) can recognize - * the conflict without re-declaring the literal. - */ - public static final String UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX = - "Failed to update the entity: "; - private static final TableMetaService INSTANCE = new TableMetaService(); private BasePOStorageOps<TablePO, TableMetaMapper> ops; @@ -123,8 +112,9 @@ public class TableMetaService { TablePO.Builder builder = TablePO.builder(); fillTablePOBuilderParentEntityId(builder, tableEntity.namespace()); - AtomicReference<TablePO> tablePORef = new AtomicReference<>(); TablePO po = POConverters.initializeTablePOWithVersion(tableEntity, builder); + // The schema lock, table row, version row, and columns share one transaction. If any later + // step fails, the earlier inserts are rolled back as well. SessionUtils.doMultipleWithCommit( // Hold the parent schema row until this transaction ends, so the table cannot be // written below a schema that is being dropped. @@ -137,11 +127,7 @@ public class TableMetaService { po.getMetalakeId()), () -> SessionUtils.doWithoutCommit( - TableMetaMapper.class, - mapper -> { - tablePORef.set(po); - ops.insertPO(mapper, po, overwrite); - }), + TableMetaMapper.class, mapper -> ops.insertPO(mapper, po, overwrite)), () -> SessionUtils.doWithoutCommit( TableVersionMapper.class, @@ -155,14 +141,12 @@ public class TableMetaService { () -> { // We need to delete the columns first if we want to overwrite the table. if (overwrite) { - TableColumnMetaService.getInstance() - .deleteColumnsByTableId(tablePORef.get().getTableId()); + TableColumnMetaService.getInstance().deleteColumnsByTableId(po.getTableId()); } }, () -> { if (tableEntity.columns() != null && !tableEntity.columns().isEmpty()) { - TableColumnMetaService.getInstance() - .insertColumnPOs(tablePORef.get(), tableEntity.columns()); + TableColumnMetaService.getInstance().insertColumnPOs(po, tableEntity.columns()); } }); @@ -200,12 +184,12 @@ public class TableMetaService { TablePO newTablePO = POConverters.updateTablePOWithVersionAndSchemaId(oldTablePO, newTableEntity, newSchemaId); - final AtomicInteger updateResult = new AtomicInteger(0); try { SessionUtils.doMultipleWithCommit( () -> { - // Only a rename that moves the table to another schema needs a lock here, and it is the - // new parent that has to stay alive, not the old one. + // Only an update that moves the table to another schema needs a lock here. The new + // parent must stay alive until the move commits; locking the old parent would not + // protect the table's new location. if (isSchemaChanged) { SchemaMetaService.getInstance() .lockSchemaForEntityWrite( @@ -215,24 +199,35 @@ public class TableMetaService { oldTablePO.getMetalakeId()); } }, - () -> - updateResult.set( - SessionUtils.getWithoutCommit( - TableMetaMapper.class, - mapper -> ops.updatePO(mapper, newTablePO, oldTablePO))), - () -> - SessionUtils.doWithoutCommit( - TableVersionMapper.class, - mapper -> { - mapper.softDeleteTableVersionByTableIdAndVersion( - oldTablePO.getTableId(), oldTablePO.getCurrentVersion()); - mapper.insertTableVersionOnDuplicateKeyUpdate(newTablePO); - }), () -> { - if (updateResult.get() > 0) { - TableColumnMetaService.getInstance() - .updateColumnPOsFromTableDiff(oldTableEntity, newTableEntity, newTablePO); + // This update is the decision point for the whole transaction. current_version is the + // table's OCC token: if another writer changed the table after we read it, that writer + // has already increased the token and this UPDATE changes zero rows. Throwing here + // rolls back the transaction before it can touch the version history or columns. + int updated = + SessionUtils.getWithoutCommit( + TableMetaMapper.class, mapper -> ops.updatePO(mapper, newTablePO, oldTablePO)); + if (updated == 0) { + throw tableWriteFailure(identifier, oldTablePO); } + }, + () -> { + // The table details live in table_version_info, while table_meta points to the current + // version. These two rows must move together. This step runs only after the table_meta + // CAS above succeeds, so a losing writer cannot overwrite the winner's version row. + SessionUtils.doWithoutCommit( + TableVersionMapper.class, + mapper -> { + mapper.softDeleteTableVersionByTableIdAndVersion( + oldTablePO.getTableId(), oldTablePO.getCurrentVersion()); + mapper.insertTableVersionOnDuplicateKeyUpdate(newTablePO); + }); + }, + () -> { + // Column changes use the same new table version. Keeping this in the same transaction + // means a column failure also rolls back table_meta and table_version_info. + TableColumnMetaService.getInstance() + .updateColumnPOsFromTableDiff(oldTableEntity, newTableEntity, newTablePO); }); } catch (RuntimeException re) { @@ -241,61 +236,42 @@ public class TableMetaService { throw re; } - if (updateResult.get() > 0) { - return newTableEntity; - } else { - throw new IOException(UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX + identifier); - } + return newTableEntity; } @Monitored(metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteTable") public boolean deleteTable(NameIdentifier identifier) { TablePO tablePO = getTablePOByIdentifier(identifier); - AtomicInteger deleteResult = new AtomicInteger(0); SessionUtils.doMultipleWithCommit( - () -> - deleteResult.set( - SessionUtils.getWithoutCommit( - TableMetaMapper.class, - mapper -> mapper.softDeleteTableMetasByTableId(tablePO.getTableId()))), () -> { - if (deleteResult.get() > 0) { - SessionUtils.doWithoutCommit( - OwnerMetaMapper.class, - mapper -> - mapper.softDeleteOwnerRelByMetadataObjectIdAndType( - tablePO.getTableId(), MetadataObject.Type.TABLE.name())); - TableColumnMetaService.getInstance().deleteColumnsByTableId(tablePO.getTableId()); - SessionUtils.doWithoutCommit( - SecurableObjectMapper.class, - mapper -> - mapper.softDeleteObjectRelsByMetadataObject( - tablePO.getTableId(), MetadataObject.Type.TABLE.name())); - SessionUtils.doWithoutCommit( - TagMetadataObjectRelMapper.class, - mapper -> - mapper.softDeleteTagMetadataObjectRelsByMetadataObject( - tablePO.getTableId(), MetadataObject.Type.TABLE.name())); - SessionUtils.doWithoutCommit( - TagMetadataObjectRelMapper.class, - mapper -> mapper.softDeleteTagMetadataObjectRelsByTableId(tablePO.getTableId())); + // Delete the table row first and only if it still has the version we read. A stale drop + // stops here, before it can remove columns, tags, policies, or any other related data. + deleteTableWithVersion(identifier, tablePO); + }, + () -> deleteTableDependents(tablePO)); - SessionUtils.doWithoutCommit( - StatisticMetaMapper.class, - mapper -> mapper.softDeleteStatisticsByEntityId(tablePO.getTableId())); - SessionUtils.doWithoutCommit( - PolicyMetadataObjectRelMapper.class, - mapper -> mapper.softDeletePolicyMetadataObjectRelsByTableId(tablePO.getTableId())); - SessionUtils.doWithoutCommit( - TableVersionMapper.class, - mapper -> - mapper.softDeleteTableVersionByTableIdAndVersion( - tablePO.getTableId(), tablePO.getCurrentVersion())); - } - }); + return true; + } - return deleteResult.get() > 0; + /** + * Deletes the table root row only when it still has the version observed by the caller. + * + * <p>This method deliberately does not start or commit a transaction. Its caller must include it + * in the same transaction as dependent-row cleanup, so a later cleanup failure restores the root + * row too. Package-private access also lets concurrency tests submit a deliberately stale + * snapshot without copying the production CAS logic. + */ + void deleteTableWithVersion(NameIdentifier identifier, TablePO observedTablePO) { + int deleted = + SessionUtils.getWithoutCommit( + TableMetaMapper.class, + mapper -> + mapper.softDeleteTableMetasByTableId( + observedTablePO.getTableId(), observedTablePO.getCurrentVersion())); + if (deleted == 0) { + throw tableWriteFailure(identifier, observedTablePO); + } } @Monitored( @@ -374,4 +350,66 @@ public class TableMetaService { builder.withCatalogId(namespacedEntityId.namespaceIds()[1]); builder.withSchemaId(namespacedEntityId.entityId()); } + + private void deleteTableDependents(TablePO tablePO) { + // The table row has already passed its version check. All cleanup below uses the same database + // transaction, so either the table and every related row are deleted together, or none are. + SessionUtils.doWithoutCommit( + OwnerMetaMapper.class, + mapper -> + mapper.softDeleteOwnerRelByMetadataObjectIdAndType( + tablePO.getTableId(), MetadataObject.Type.TABLE.name())); + TableColumnMetaService.getInstance().deleteColumnsByTableId(tablePO.getTableId()); + SessionUtils.doWithoutCommit( + SecurableObjectMapper.class, + mapper -> + mapper.softDeleteObjectRelsByMetadataObject( + tablePO.getTableId(), MetadataObject.Type.TABLE.name())); + SessionUtils.doWithoutCommit( + TagMetadataObjectRelMapper.class, + mapper -> + mapper.softDeleteTagMetadataObjectRelsByMetadataObject( + tablePO.getTableId(), MetadataObject.Type.TABLE.name())); + SessionUtils.doWithoutCommit( + TagMetadataObjectRelMapper.class, + mapper -> mapper.softDeleteTagMetadataObjectRelsByTableId(tablePO.getTableId())); + SessionUtils.doWithoutCommit( + StatisticMetaMapper.class, + mapper -> mapper.softDeleteStatisticsByEntityId(tablePO.getTableId())); + SessionUtils.doWithoutCommit( + PolicyMetadataObjectRelMapper.class, + mapper -> mapper.softDeletePolicyMetadataObjectRelsByTableId(tablePO.getTableId())); + SessionUtils.doWithoutCommit( + TableVersionMapper.class, + mapper -> + mapper.softDeleteTableVersionByTableIdAndVersion( + tablePO.getTableId(), tablePO.getCurrentVersion())); + } + + private RuntimeException tableWriteFailure(NameIdentifier identifier, TablePO observedTablePO) { + // A zero-row CAS has two different meanings: + // 1. The same table is still here, but another writer changed its version. The caller should + // retry, so return OptimisticLockException. + // 2. The table ID was deleted, renamed, or moved away from the requested name. From the + // caller's point of view the requested table no longer exists, so return NoSuchEntity. + // + // Read by the stable table ID and lock the row. The lock waits for an in-flight writer to + // finish, which lets us classify the failure using committed data instead of guessing while + // the other transaction is still running. + TablePO currentTablePO = + SessionUtils.getWithoutCommit( + TableMetaMapper.class, + mapper -> mapper.selectTableMetaByIdForUpdate(observedTablePO.getTableId())); + if (currentTablePO == null + || !Objects.equals(currentTablePO.getTableName(), observedTablePO.getTableName()) + || !Objects.equals(currentTablePO.getSchemaId(), observedTablePO.getSchemaId()) + || !Objects.equals(currentTablePO.getCatalogId(), observedTablePO.getCatalogId()) + || !Objects.equals(currentTablePO.getMetalakeId(), observedTablePO.getMetalakeId())) { + return new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.TABLE.name().toLowerCase(), + identifier.name()); + } + return ExceptionUtils.concurrentModification(Entity.EntityType.TABLE, identifier); + } } 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 a834398c71..0c2eb8c091 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java @@ -58,6 +58,7 @@ import org.apache.gravitino.TestColumn; import org.apache.gravitino.auth.AuthConstants; import org.apache.gravitino.connector.TestCatalogOperations; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.lock.LockManager; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.ColumnEntity; @@ -488,7 +489,16 @@ public class TestTableOperationDispatcher extends TestOperationDispatcher { Assertions.assertEquals("test", alteredTable3.auditInfo().creator()); Assertions.assertEquals("test", alteredTable3.auditInfo().lastModifier()); - // Case 4: Test if the table entity is not matched + // Case 4: Optimistic-lock conflicts must reach the caller so it can retry. + reset(entityStore); + doThrow(new OptimisticLockException("mock conflict")) + .when(entityStore) + .update(any(), any(), any(), any()); + Assertions.assertThrows( + OptimisticLockException.class, + () -> tableOperationDispatcher.alterTable(tableIdent, changes)); + + // Case 5: Test if the table entity is not matched reset(entityStore); TableEntity unmatchedEntity = TableEntity.builder() diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestTableMetaBaseSQLProvider.java b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestTableMetaBaseSQLProvider.java new file mode 100644 index 0000000000..e3789d6be4 --- /dev/null +++ b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestTableMetaBaseSQLProvider.java @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.mapper.provider.base; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class TestTableMetaBaseSQLProvider { + + private static final TableMetaBaseSQLProvider PROVIDER = new TableMetaBaseSQLProvider(); + + @Test + void testUpdateUsesOnlyIdVersionAndActiveStateForCas() { + String sql = PROVIDER.updateTableMeta(null, null, null); + String whereClause = sql.substring(sql.indexOf(" WHERE")); + + Assertions.assertEquals( + " WHERE table_id = #{oldTableMeta.tableId}" + + " AND current_version = #{oldTableMeta.currentVersion}" + + " AND deleted_at = 0", + whereClause); + } + + @Test + void testDirectDeleteUsesVersionCas() { + String sql = PROVIDER.softDeleteTableMetasByTableId(null, null); + + Assertions.assertTrue(sql.contains("AND current_version = #{currentVersion}")); + Assertions.assertTrue(sql.endsWith("AND deleted_at = 0")); + } + + @Test + void testConflictReadLocksActiveRow() { + String sql = PROVIDER.selectTableMetaByIdForUpdate(null); + + Assertions.assertTrue(sql.contains("WHERE table_id = #{tableId} AND deleted_at = 0")); + Assertions.assertTrue(sql.endsWith("FOR UPDATE")); + } +} diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestTableMetaPostgreSQLProvider.java b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestTableMetaPostgreSQLProvider.java new file mode 100644 index 0000000000..cad40453aa --- /dev/null +++ b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestTableMetaPostgreSQLProvider.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.mapper.provider.postgresql; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class TestTableMetaPostgreSQLProvider { + + @Test + void testDirectDeleteUsesVersionCas() { + String sql = new TableMetaPostgreSQLProvider().softDeleteTableMetasByTableId(null, null); + + Assertions.assertTrue(sql.contains("AND current_version = #{currentVersion}")); + Assertions.assertTrue(sql.endsWith("AND deleted_at = 0")); + } +} diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java index a795168a5d..623a45029b 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java @@ -27,6 +27,12 @@ import java.io.IOException; import java.time.Instant; import java.util.List; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.function.Function; import java.util.stream.Collectors; import org.apache.gravitino.Entity; @@ -34,6 +40,7 @@ import org.apache.gravitino.EntityAlreadyExistsException; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.BaseMetalake; import org.apache.gravitino.meta.CatalogEntity; import org.apache.gravitino.meta.ColumnEntity; @@ -56,6 +63,10 @@ import org.apache.gravitino.rel.types.Types; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper; +import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper; +import org.apache.gravitino.storage.relational.mapper.TableMetaMapper; +import org.apache.gravitino.storage.relational.po.SchemaPO; +import org.apache.gravitino.storage.relational.po.TablePO; import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord; import org.apache.gravitino.storage.relational.po.cache.OperateType; import org.apache.gravitino.storage.relational.utils.SessionUtils; @@ -101,6 +112,204 @@ public class TestTableMetaService extends TestJDBCBackend { assertThrows(EntityAlreadyExistsException.class, () -> backend.insert(tableCopy, false)); } + @TestTemplate + public void testInsertWaitsForConcurrentSchemaDelete() throws Exception { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity schema = createAndInsertSchema(metalakeName, catalogName, schemaName); + SchemaPO observedSchemaPO = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(schema.id())); + TableEntity table = + createTableEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofTable(metalakeName, catalogName, schemaName), + "table_racing_schema_delete", + AUDIT_INFO); + + CountDownLatch schemaDeleteLocked = new CountDownLatch(1); + CountDownLatch allowDeleteCommit = new CountDownLatch(1); + CountDownLatch tableInsertStarted = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(2); + Future<Throwable> deleteResult = + executor.submit( + () -> { + try { + SessionUtils.doMultipleWithCommit( + () -> { + int deleted = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + mapper.softDeleteSchemaMetaBySchemaIdAndVersion( + observedSchemaPO.getSchemaId(), + observedSchemaPO.getCurrentVersion())); + Assertions.assertEquals(1, deleted); + schemaDeleteLocked.countDown(); + try { + assertTrue(allowDeleteCommit.await(30, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + }); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + try { + assertTrue(schemaDeleteLocked.await(30, TimeUnit.SECONDS)); + Future<Throwable> insertResult = + executor.submit( + () -> { + tableInsertStarted.countDown(); + try { + TableMetaService.getInstance().insertTable(table, false); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + assertTrue(tableInsertStarted.await(30, TimeUnit.SECONDS)); + assertThrows(TimeoutException.class, () -> insertResult.get(500, TimeUnit.MILLISECONDS)); + + allowDeleteCommit.countDown(); + Assertions.assertNull(deleteResult.get(30, TimeUnit.SECONDS)); + Assertions.assertInstanceOf( + NoSuchEntityException.class, insertResult.get(30, TimeUnit.SECONDS)); + Assertions.assertTrue( + SessionUtils.getWithoutCommit( + TableMetaMapper.class, + mapper -> mapper.listTablePOsByTableIds(List.of(table.id()))) + .isEmpty()); + } finally { + allowDeleteCommit.countDown(); + executor.shutdownNow(); + } + } + + @TestTemplate + public void testInsertRollsBackAllRowsWhenColumnWriteFails() throws IOException { + createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO); + Namespace tableNamespace = NamespaceUtil.ofTable(metalakeName, catalogName, schemaName); + ColumnEntity column = + ColumnEntity.builder() + .withId(RandomIdGenerator.INSTANCE.nextId()) + .withName("column") + .withPosition(0) + .withDataType(Types.IntegerType.get()) + .withNullable(true) + .withAutoIncrement(false) + .withAuditInfo(AUDIT_INFO) + .build(); + TableEntity invalidTable = + TableEntity.builder() + .withId(RandomIdGenerator.INSTANCE.nextId()) + .withName("table_insert_rollback") + .withNamespace(tableNamespace) + // The duplicate ID violates the column-version unique key. The failure happens after + // table_meta and table_version_info have already been written in this transaction. + .withColumns(List.of(column, column)) + .withAuditInfo(AUDIT_INFO) + .build(); + + assertThrows( + RuntimeException.class, + () -> TableMetaService.getInstance().insertTable(invalidTable, false)); + assertThrows( + NoSuchEntityException.class, + () -> TableMetaService.getInstance().getTableByIdentifier(invalidTable.nameIdentifier())); + + // Reusing the same table ID, name, and column proves that the failed attempt left no metadata, + // version, or column row behind. + TableEntity validTable = + TableEntity.builder() + .withId(invalidTable.id()) + .withName(invalidTable.name()) + .withNamespace(tableNamespace) + .withColumns(List.of(column)) + .withAuditInfo(AUDIT_INFO) + .build(); + TableMetaService.getInstance().insertTable(validTable, false); + TableEntity inserted = + TableMetaService.getInstance().getTableByIdentifier(validTable.nameIdentifier()); + Assertions.assertEquals(validTable.id(), inserted.id()); + Assertions.assertEquals(1, inserted.columns().size()); + } + + @TestTemplate + public void testOverwriteRollsBackExistingTableAndThenSucceeds() throws IOException { + createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO); + Namespace tableNamespace = NamespaceUtil.ofTable(metalakeName, catalogName, schemaName); + ColumnEntity originalColumn = + ColumnEntity.builder() + .withId(RandomIdGenerator.INSTANCE.nextId()) + .withName("original_column") + .withPosition(0) + .withDataType(Types.IntegerType.get()) + .withNullable(true) + .withAutoIncrement(false) + .withAuditInfo(AUDIT_INFO) + .build(); + TableEntity original = + TableEntity.builder() + .withId(RandomIdGenerator.INSTANCE.nextId()) + .withName("table_overwrite_rollback") + .withNamespace(tableNamespace) + .withColumns(List.of(originalColumn)) + .withComment("original") + .withAuditInfo(AUDIT_INFO) + .build(); + TableMetaService.getInstance().insertTable(original, false); + + ColumnEntity replacementColumn = + ColumnEntity.builder() + .withId(RandomIdGenerator.INSTANCE.nextId()) + .withName("replacement_column") + .withPosition(0) + .withDataType(Types.StringType.get()) + .withNullable(true) + .withAutoIncrement(false) + .withAuditInfo(AUDIT_INFO) + .build(); + TableEntity invalidReplacement = + TableEntity.builder() + .withId(original.id()) + .withName(original.name()) + .withNamespace(tableNamespace) + // Overwrite deletes the old columns before inserting these. Repeating the same ID + // makes the final step fail, which must restore both the table and its old column. + .withColumns(List.of(replacementColumn, replacementColumn)) + .withComment("must roll back") + .withAuditInfo(AUDIT_INFO) + .build(); + assertThrows( + RuntimeException.class, + () -> TableMetaService.getInstance().insertTable(invalidReplacement, true)); + + TableEntity afterFailure = + TableMetaService.getInstance().getTableByIdentifier(original.nameIdentifier()); + Assertions.assertEquals("original", afterFailure.comment()); + Assertions.assertEquals(List.of(originalColumn), afterFailure.columns()); + + TableEntity validReplacement = + TableEntity.builder() + .withId(original.id()) + .withName(original.name()) + .withNamespace(tableNamespace) + .withColumns(List.of(replacementColumn)) + .withComment("replaced") + .withAuditInfo(AUDIT_INFO) + .build(); + TableMetaService.getInstance().insertTable(validReplacement, true); + + TableEntity afterSuccess = + TableMetaService.getInstance().getTableByIdentifier(original.nameIdentifier()); + Assertions.assertEquals("replaced", afterSuccess.comment()); + Assertions.assertEquals(List.of(replacementColumn), afterSuccess.columns()); + } + @TestTemplate public void testUpdateAlreadyExistsException() throws IOException { createAndInsertMakeLake(metalakeName); @@ -318,6 +527,273 @@ public class TestTableMetaService extends TestJDBCBackend { && record.getOperateType() == OperateType.DROP)); } + @TestTemplate + public void testAlterReportsOptimisticLockConflictAndKeepsWinnerVersion() throws IOException { + createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO); + TableEntity table = + createTableEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofTable(metalakeName, catalogName, schemaName), + "table_alter_conflict", + AUDIT_INFO); + backend.insert(table, false); + TablePO initialPO = getTablePO(table.id()); + + assertThrows( + OptimisticLockException.class, + () -> + TableMetaService.getInstance() + .updateTable( + table.nameIdentifier(), + entity -> { + TableEntity current = (TableEntity) entity; + // The updater runs before the outer CAS. Commit another update here so the + // outer write continues with a stale current_version and must lose the CAS. + updateTableUnchecked( + table.nameIdentifier(), + competing -> copyTableWithComment(competing, "competing update")); + return copyTableWithComment(current, "requested update"); + })); + + TableEntity current = + TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier()); + Assertions.assertEquals("competing update", current.comment()); + TablePO currentPO = getTablePO(table.id()); + Assertions.assertEquals( + initialPO.getCurrentVersion() + 1, currentPO.getCurrentVersion().longValue()); + Assertions.assertEquals(currentPO.getCurrentVersion(), currentPO.getLastVersion()); + } + + @TestTemplate + public void testAlterReportsNoSuchWhenDeletedConcurrently() throws IOException { + createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO); + TableEntity table = + createTableEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofTable(metalakeName, catalogName, schemaName), + "table_alter_deleted", + AUDIT_INFO); + backend.insert(table, false); + + assertThrows( + NoSuchEntityException.class, + () -> + TableMetaService.getInstance() + .updateTable( + table.nameIdentifier(), + entity -> { + TableMetaService.getInstance().deleteTable(table.nameIdentifier()); + return copyTableWithComment((TableEntity) entity, "requested update"); + })); + assertThrows( + NoSuchEntityException.class, + () -> TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier())); + } + + @TestTemplate + public void testAlterReportsNoSuchWhenRenamedConcurrently() throws IOException { + createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO); + TableEntity table = + createTableEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofTable(metalakeName, catalogName, schemaName), + "table_alter_renamed", + AUDIT_INFO); + backend.insert(table, false); + NameIdentifier renamedIdentifier = + NameIdentifier.of(table.namespace(), "table_alter_renamed_winner"); + + assertThrows( + NoSuchEntityException.class, + () -> + TableMetaService.getInstance() + .updateTable( + table.nameIdentifier(), + entity -> { + TableEntity current = (TableEntity) entity; + updateTableUnchecked( + table.nameIdentifier(), + competing -> + copyTable( + competing, + renamedIdentifier.name(), + competing.namespace(), + "renamed winner")); + return copyTableWithComment(current, "stale update"); + })); + + assertThrows( + NoSuchEntityException.class, + () -> TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier())); + Assertions.assertEquals( + "renamed winner", + TableMetaService.getInstance().getTableByIdentifier(renamedIdentifier).comment()); + } + + @TestTemplate + public void testAlterReportsNoSuchWhenMovedConcurrently() throws IOException { + createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO); + String newSchemaName = "schema_for_concurrent_move"; + createAndInsertSchema(metalakeName, catalogName, newSchemaName); + TableEntity table = + createTableEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofTable(metalakeName, catalogName, schemaName), + "table_alter_moved", + AUDIT_INFO); + backend.insert(table, false); + Namespace movedNamespace = NamespaceUtil.ofTable(metalakeName, catalogName, newSchemaName); + NameIdentifier movedIdentifier = NameIdentifier.of(movedNamespace, table.name()); + + assertThrows( + NoSuchEntityException.class, + () -> + TableMetaService.getInstance() + .updateTable( + table.nameIdentifier(), + entity -> { + TableEntity current = (TableEntity) entity; + updateTableUnchecked( + table.nameIdentifier(), + competing -> + copyTable( + competing, competing.name(), movedNamespace, "moved winner")); + return copyTableWithComment(current, "stale update"); + })); + + assertThrows( + NoSuchEntityException.class, + () -> TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier())); + Assertions.assertEquals( + "moved winner", + TableMetaService.getInstance().getTableByIdentifier(movedIdentifier).comment()); + } + + @TestTemplate + public void testDeleteRejectsStaleVersion() throws IOException { + createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO); + ColumnEntity column = + ColumnEntity.builder() + .withId(RandomIdGenerator.INSTANCE.nextId()) + .withName("column_that_must_survive") + .withPosition(0) + .withDataType(Types.IntegerType.get()) + .withNullable(true) + .withAutoIncrement(false) + .withAuditInfo(AUDIT_INFO) + .build(); + TableEntity table = + TableEntity.builder() + .withId(RandomIdGenerator.INSTANCE.nextId()) + .withName("table_stale_delete") + .withNamespace(NamespaceUtil.ofTable(metalakeName, catalogName, schemaName)) + .withColumns(List.of(column)) + .withAuditInfo(AUDIT_INFO) + .build(); + backend.insert(table, false); + TablePO stalePO = getTablePO(table.id()); + + TableMetaService.getInstance() + .updateTable( + table.nameIdentifier(), + entity -> copyTableWithComment((TableEntity) entity, "winning update")); + + // The stale drop still carries the original version. It must not delete the newer table. + assertThrows( + OptimisticLockException.class, + () -> + SessionUtils.doMultipleWithCommit( + () -> + TableMetaService.getInstance() + .deleteTableWithVersion(table.nameIdentifier(), stalePO))); + TableEntity current = + TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier()); + Assertions.assertEquals("winning update", current.comment()); + Assertions.assertEquals(1, current.columns().size()); + Assertions.assertEquals(column.id(), current.columns().get(0).id()); + } + + @TestTemplate + public void testDeleteReportsNoSuchWhenDeletedConcurrently() throws IOException { + createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO); + TableEntity table = + createTableEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofTable(metalakeName, catalogName, schemaName), + "table_delete_deleted", + AUDIT_INFO); + backend.insert(table, false); + TablePO stalePO = getTablePO(table.id()); + + TableMetaService.getInstance().deleteTable(table.nameIdentifier()); + + // The second delete still has the first delete's snapshot. Since the table is now gone rather + // than merely newer, the result must be "not found", not an OCC conflict. + assertThrows( + NoSuchEntityException.class, + () -> + SessionUtils.doMultipleWithCommit( + () -> + TableMetaService.getInstance() + .deleteTableWithVersion(table.nameIdentifier(), stalePO))); + } + + @TestTemplate + public void testUpdateRollsBackMetadataAndVersionWhenColumnWriteFails() throws IOException { + createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO); + TableEntity table = + createTableEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofTable(metalakeName, catalogName, schemaName), + "table_update_rollback", + AUDIT_INFO); + backend.insert(table, false); + TablePO initialPO = getTablePO(table.id()); + + ColumnEntity duplicateColumn = + ColumnEntity.builder() + .withId(RandomIdGenerator.INSTANCE.nextId()) + .withName("duplicate_id") + .withPosition(0) + .withDataType(Types.IntegerType.get()) + .withNullable(true) + .withAutoIncrement(false) + .withAuditInfo(AUDIT_INFO) + .build(); + // The duplicate column ID fails after the table and version rows are updated. The assertions + // below verify that the outer transaction rolls both earlier writes back. + assertThrows( + IllegalStateException.class, + () -> + TableMetaService.getInstance() + .updateTable( + table.nameIdentifier(), + entity -> { + TableEntity current = (TableEntity) entity; + return TableEntity.builder() + .withId(current.id()) + .withName(current.name()) + .withNamespace(current.namespace()) + .withColumns(List.of(duplicateColumn, duplicateColumn)) + .withProperties(current.properties()) + .withPartitioning(current.partitioning()) + .withSortOrders(current.sortOrders()) + .withDistribution(current.distribution()) + .withIndexes(current.indexes()) + .withComment("must roll back") + .withAuditInfo(current.auditInfo()) + .build(); + })); + + TableEntity current = + TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier()); + Assertions.assertNull(current.comment()); + Assertions.assertTrue(current.columns().isEmpty()); + TablePO currentPO = getTablePO(table.id()); + Assertions.assertEquals(initialPO.getCurrentVersion(), currentPO.getCurrentVersion()); + Assertions.assertEquals(initialPO.getLastVersion(), currentPO.getLastVersion()); + } + @TestTemplate public void testBatchGetTableByIdentifierIncludesVersionInfoFields() throws IOException { createAndInsertMakeLake(metalakeName); @@ -449,4 +925,39 @@ public class TestTableMetaService extends TestJDBCBackend { Assertions.assertEquals(expectedColumn.auditInfo(), column.auditInfo()); }); } + + private TablePO getTablePO(long tableId) { + return SessionUtils.getWithoutCommit( + TableMetaMapper.class, mapper -> mapper.listTablePOsByTableIds(List.of(tableId)).get(0)); + } + + private TableEntity copyTableWithComment(TableEntity current, String comment) { + return copyTable(current, current.name(), current.namespace(), comment); + } + + private TableEntity copyTable( + TableEntity current, String name, Namespace namespace, String comment) { + return TableEntity.builder() + .withId(current.id()) + .withName(name) + .withNamespace(namespace) + .withColumns(current.columns()) + .withProperties(current.properties()) + .withPartitioning(current.partitioning()) + .withSortOrders(current.sortOrders()) + .withDistribution(current.distribution()) + .withIndexes(current.indexes()) + .withComment(comment) + .withAuditInfo(current.auditInfo()) + .build(); + } + + private void updateTableUnchecked( + NameIdentifier identifier, Function<TableEntity, TableEntity> updater) { + try { + TableMetaService.getInstance().updateTable(identifier, updater); + } catch (IOException e) { + throw new RuntimeException(e); + } + } }
