This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 4e74534ae0 [#12345] improvement(core): add OCC for table writes
(#12551)
4e74534ae0 is described below
commit 4e74534ae053292420dc45b102e21abe613c09c2
Author: Qi Yu <[email protected]>
AuthorDate: Wed Aug 26 16:38:41 2026 +0800
[#12345] improvement(core): add OCC for table writes (#12551)
### What changes were proposed in this pull request?
- Use `current_version` as the OCC token for table alter and direct
delete.
- Execute the root table CAS before version, column, and dependent-row
writes in one transaction.
- Distinguish a concurrent version change from a deleted, renamed, or
moved table.
- Let optimistic-lock conflicts reach the caller on the managed paths,
which write to the store directly. The best-effort
`OperationDispatcher.operateOnEntity` helper keeps swallowing them: only
external entities reach it, and there the catalog was already changed,
so failing would invite a retry that re-applies a non-idempotent change.
- Retry only optimistic-lock conflicts for idempotent Lance metadata
repair.
- Add comments explaining the transaction ordering, CAS predicates,
conflict classification, and retry behavior.
- Add comprehensive unit and cross-database tests.
This PR depends on #12456.
Two things are deliberately left out. The conflict classification
(`tableWriteFailure`) and the locking select now exist once per entity
type; extracting them is a follow-up once the series is done, because
what differs per entity is the mapper, which identity columns to
compare, and schema's physical-to-logical conversion, so doing it now
means refactoring three merged services from inside a fourth. And a
version conflict while dropping is tracked in #12597, which covers every
entity type rather than this path alone.
### Why are the changes needed?
Concurrent table writes could overwrite the winning version metadata,
while a stale delete could remove data belonging to a newer table
version. Some optimistic-lock conflicts were also swallowed or treated
as generic IO failures.
Fix: #12345
### Does this PR introduce _any_ user-facing change?
Stale managed-table writes now fail with the existing optimistic-lock
conflict response instead of silently overwriting newer metadata. Reads
and the import path are unchanged: loading an entity still repairs the
Gravitino copy on a best-effort basis, so a load that loses a version
race keeps returning the entity. No API or property is added.
### How was this patch tested?
- Ran 48 `TableMetaService` cases across H2, MySQL, and PostgreSQL.
- Ran all 16 `TestTableOperationDispatcher` tests.
- Ran 4 SQL provider OCC tests.
- Ran 27 Lance table and concurrent-repair tests.
- Ran Spotless and `git diff --check`.
---
.../lakehouse/lance/LanceTableOperations.java | 40 +-
.../lance/TestLanceConcurrentRepairStress.java | 19 +-
.../lakehouse/lance/TestLanceTableOperations.java | 94 ++-
.../gravitino/catalog/OperationDispatcher.java | 19 +
.../catalog/TableOperationDispatcher.java | 5 +
.../storage/relational/mapper/TableMetaMapper.java | 19 +-
.../mapper/TableMetaSQLProviderFactory.java | 22 +-
.../provider/base/TableMetaBaseSQLProvider.java | 63 +-
.../postgresql/TableMetaPostgreSQLProvider.java | 15 +-
.../gravitino/storage/relational/po/TablePO.java | 33 ++
.../relational/service/TableMetaService.java | 243 +++++---
.../catalog/TestTableOperationDispatcher.java | 53 +-
.../base/TestTableMetaBaseSQLProvider.java | 66 +++
.../TestTableMetaPostgreSQLProvider.java | 61 ++
.../storage/relational/po/TestTablePO.java | 88 +++
.../relational/service/TestTableMetaService.java | 657 ++++++++++++++++++---
16 files changed, 1280 insertions(+), 217 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..87e2bcb93b 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;
@@ -214,11 +215,29 @@ public abstract class OperationDispatcher {
}
}
+ /**
+ * Runs a store operation as a best-effort side effect of the request.
+ *
+ * <p>Every failure is logged and reported as a null result, because the
external catalog is the
+ * source of truth on these paths: a load that imports or repairs the
Gravitino copy must still
+ * return the entity it read, and the next load repairs what this one could
not write.
+ */
protected <R extends HasIdentifier> R operateOnEntity(
NameIdentifier ident, ThrowableFunction<NameIdentifier, R> fn, String
opName, long id) {
R ret = null;
try {
ret = fn.apply(ident);
+ } catch (OptimisticLockException e) {
+ // Only external entities reach this point, so swallowing the conflict
is safe: alterTable,
+ // alterSchema and alterView return before calling this helper when the
entity is managed,
+ // and no catalog reports managed storage for topics
(KafkaCatalogCapability). A managed
+ // alter therefore hits the store directly and its conflict still
reaches the caller.
+ //
+ // For an external entity the catalog was already changed and remains
the source of truth.
+ // Failing the request would invite a retry that re-applies the external
change, and some
+ // changes are not idempotent, so the stale Gravitino copy is the lesser
problem: the next
+ // load imports the entity again.
+ LOG.warn(FormattedErrorMessages.STORE_OP_FAILURE, opName, ident, 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/catalog/TableOperationDispatcher.java
b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
index a53722c4f4..5f7c18f53c 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
@@ -51,6 +51,7 @@ import
org.apache.gravitino.exceptions.GravitinoRuntimeException;
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.lock.LockType;
import org.apache.gravitino.lock.TreeLockUtils;
@@ -416,6 +417,8 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
if (droppedFromCatalog) {
try {
store.delete(ident, TABLE);
+ } catch (OptimisticLockException e) {
+ throw e;
} catch (NoSuchEntityException e) {
LOG.warn("The table to be dropped does not exist in the store:
{}", ident, e);
} catch (Exception e) {
@@ -470,6 +473,8 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
if (droppedFromCatalog) {
try {
store.delete(ident, TABLE);
+ } catch (OptimisticLockException e) {
+ throw e;
} catch (NoSuchEntityException e) {
LOG.warn("The table to be purged does not exist in the store:
{}", ident, e);
} catch (Exception e) {
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..bb2db5eb33 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,29 @@ 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.
+ *
+ * <p>Unlike the metalake, catalog and schema providers, this cannot be
written as {@code
+ * selectTableMetaById(id) + " FOR UPDATE"}: that select LEFT JOINs {@code
table_version_info},
+ * and locking the nullable side of an outer join is rejected by PostgreSQL
and locks the wrong
+ * rows on MySQL. The projection is therefore spelled out for {@code
table_meta} alone, and the
+ * returned row carries only the identity and version columns its callers
read.
+ *
+ * @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
@@ -220,11 +243,26 @@ public class TableMetaBaseSQLProvider {
+ " catalog_id = #{tableMeta.catalogId},"
+ " schema_id = #{tableMeta.schemaId},"
+ " audit_info = #{tableMeta.auditInfo},"
- + " current_version = #{tableMeta.currentVersion},"
- + " last_version = #{tableMeta.lastVersion},"
+ // Keep the OCC token monotonic on overwrite. Resetting it to the
incoming initial version
+ // would let a writer that read the same old version pass its CAS
after this statement.
+ // last_version is assigned first, so both columns advance from the
stored current version.
+ + " last_version = current_version + 1,"
+ + " current_version = current_version + 1,"
+ " 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 +278,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..a3f4c53ac7 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
@@ -50,17 +50,24 @@ public class TableMetaPostgreSQLProvider extends
TableMetaBaseSQLProvider {
+ " catalog_id = #{tableMeta.catalogId},"
+ " schema_id = #{tableMeta.schemaId},"
+ " audit_info = #{tableMeta.auditInfo},"
- + " current_version = #{tableMeta.currentVersion},"
- + " last_version = #{tableMeta.lastVersion},"
+ // Qualify the stored row's version because a bare name is ambiguous
on this side of
+ // ON CONFLICT in PostgreSQL.
+ + " current_version = "
+ + TABLE_NAME
+ + ".current_version + 1,"
+ + " last_version = "
+ + TABLE_NAME
+ + ".current_version + 1,"
+ " deleted_at = #{tableMeta.deletedAt}";
}
@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/po/TablePO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/TablePO.java
index 56fea38337..22d92a76b9 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/po/TablePO.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/po/TablePO.java
@@ -120,6 +120,28 @@ public class TablePO {
tablePO = new TablePO();
}
+ // Copies every column of the source row. When a field is added to TablePO
above, add it here
+ // too, otherwise callers that copy a row would silently blank it.
+ private Builder(TablePO source) {
+ tablePO = new TablePO();
+ tablePO.tableId = source.tableId;
+ tablePO.tableName = source.tableName;
+ tablePO.metalakeId = source.metalakeId;
+ tablePO.catalogId = source.catalogId;
+ tablePO.schemaId = source.schemaId;
+ tablePO.auditInfo = source.auditInfo;
+ tablePO.currentVersion = source.currentVersion;
+ tablePO.lastVersion = source.lastVersion;
+ tablePO.deletedAt = source.deletedAt;
+ tablePO.format = source.format;
+ tablePO.properties = source.properties;
+ tablePO.partitions = source.partitions;
+ tablePO.sortOrders = source.sortOrders;
+ tablePO.distribution = source.distribution;
+ tablePO.indexes = source.indexes;
+ tablePO.comment = source.comment;
+ }
+
public Builder withTableId(Long tableId) {
tablePO.tableId = tableId;
return this;
@@ -226,4 +248,15 @@ public class TablePO {
public static Builder builder() {
return new Builder();
}
+
+ /**
+ * Creates a new instance of {@link Builder} pre-filled with the values of
an existing table, so
+ * that a caller changing a few columns does not have to restate the others.
+ *
+ * @param tablePO The table to copy the values from.
+ * @return The new instance.
+ */
+ public static Builder builder(TablePO tablePO) {
+ return new Builder(tablePO);
+ }
}
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..a387bc370f 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,7 +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;
@@ -55,15 +54,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 +113,10 @@ public class TableMetaService {
TablePO.Builder builder = TablePO.builder();
fillTablePOBuilderParentEntityId(builder, tableEntity.namespace());
- AtomicReference<TablePO> tablePORef = new AtomicReference<>();
TablePO po = POConverters.initializeTablePOWithVersion(tableEntity,
builder);
+ AtomicReference<TablePO> persistedPO = new AtomicReference<>(po);
+ // 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.
@@ -139,15 +131,38 @@ public class TableMetaService {
SessionUtils.doWithoutCommit(
TableMetaMapper.class,
mapper -> {
- tablePORef.set(po);
ops.insertPO(mapper, po, overwrite);
+ if (overwrite) {
+ // MySQL may resolve the upsert through the active
(schema_id, table_name,
+ // deleted_at) key rather than table_id. In that case it
preserves the
+ // winner's ID. The upsert already holds that row until
commit, so read the
+ // database-derived identity and version back through
the same natural key.
+ TablePO storedPO =
+ mapper.selectTableMetaBySchemaIdAndName(
+ po.getSchemaId(), po.getTableName());
+ Preconditions.checkState(
+ storedPO != null,
+ "The overwritten table %s in schema %s does not
exist",
+ po.getTableName(),
+ po.getSchemaId());
+
persistedPO.set(tablePOWithPersistedIdentityAndVersions(po, storedPO));
+ }
}),
() ->
SessionUtils.doWithoutCommit(
TableVersionMapper.class,
mapper -> {
if (overwrite) {
- mapper.insertTableVersionOnDuplicateKeyUpdate(po);
+ TablePO storedPO = persistedPO.get();
+ // Retire the version row this overwrite replaces. There
is one only when the
+ // upsert updated an existing table: the database then
moved the version from
+ // N to N + 1, so the row to retire is N. When the
upsert inserted a brand new
+ // table the version is still the initial one and no
earlier row exists.
+ if (storedPO.getCurrentVersion() >
POConverters.INIT_VERSION) {
+ mapper.softDeleteTableVersionByTableIdAndVersion(
+ storedPO.getTableId(),
storedPO.getCurrentVersion() - 1);
+ }
+ mapper.insertTableVersionOnDuplicateKeyUpdate(storedPO);
} else {
mapper.insertTableVersion(po);
}
@@ -156,13 +171,13 @@ 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());
+ .deleteColumnsByTableId(persistedPO.get().getTableId());
}
},
() -> {
if (tableEntity.columns() != null &&
!tableEntity.columns().isEmpty()) {
TableColumnMetaService.getInstance()
- .insertColumnPOs(tablePORef.get(), tableEntity.columns());
+ .insertColumnPOs(persistedPO.get(), tableEntity.columns());
}
});
@@ -200,12 +215,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 +230,42 @@ 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, keyed by
(table_id, version), while
+ // table_meta only points at the current version. The two rows
have to move together,
+ // and the upsert below has no version guard of its own: it
overwrites whatever sits
+ // under that key.
+ //
+ // Say two writers both read version 5 and both want to write 6.
Their version rows
+ // carry the same key, (table_id, 6), so whichever runs this
statement second would
+ // silently replace the other's details. Ordering this step after
the table_meta CAS is
+ // what prevents that: the loser matches no row up there, throws,
and the transaction
+ // rolls back before reaching this statement. Only the winner ever
writes version 6.
+ 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 +274,39 @@ 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);
+ // Delete the table row first and only if it still has the version we
read. A stale drop stops
+ // there, before it can remove columns, tags, policies, or any other
related data.
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()));
+ () -> 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 +385,76 @@ public class TableMetaService {
builder.withCatalogId(namespacedEntityId.namespaceIds()[1]);
builder.withSchemaId(namespacedEntityId.entityId());
}
+
+ private TablePO tablePOWithPersistedIdentityAndVersions(TablePO incomingPO,
TablePO persistedPO) {
+ // The upsert derives the version inside the database and may preserve an
existing table ID, so
+ // its dependent rows must carry the identity and versions the database
ended up with.
+ return TablePO.builder(incomingPO)
+ .withTableId(persistedPO.getTableId())
+ .withCurrentVersion(persistedPO.getCurrentVersion())
+ .withLastVersion(persistedPO.getLastVersion())
+ .build();
+ }
+
+ 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 0bf03cfb06..faf1dd83ab 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
@@ -60,6 +60,7 @@ import org.apache.gravitino.connector.TestCatalogOperations;
import org.apache.gravitino.exceptions.GravitinoRuntimeException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchTableException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.ColumnEntity;
@@ -490,7 +491,17 @@ 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: The external alter has already succeeded, so an internal mirror
conflict is
+ // best-effort. Returning an error here could make the client apply the
external change twice.
+ reset(entityStore);
+ doThrow(new OptimisticLockException("mock conflict"))
+ .when(entityStore)
+ .update(any(), any(), any(), any());
+ Table alteredTable4 = tableOperationDispatcher.alterTable(tableIdent,
changes);
+ Assertions.assertEquals("test", alteredTable4.auditInfo().creator());
+ Assertions.assertEquals("test", alteredTable4.auditInfo().lastModifier());
+
+ // Case 5: Test if the table entity is not matched.
reset(entityStore);
TableEntity unmatchedEntity =
TableEntity.builder()
@@ -501,10 +512,10 @@ public class TestTableOperationDispatcher extends
TestOperationDispatcher {
AuditInfo.builder().withCreator("gravitino").withCreateTime(Instant.now()).build())
.build();
doReturn(unmatchedEntity).when(entityStore).update(any(), any(), any(),
any());
- Table alteredTable4 = tableOperationDispatcher.alterTable(tableIdent,
changes);
+ Table alteredTable5 = tableOperationDispatcher.alterTable(tableIdent,
changes);
// Audit info is gotten from the catalog, not from the entity store
- Assertions.assertEquals("test", alteredTable4.auditInfo().creator());
- Assertions.assertEquals("test", alteredTable4.auditInfo().lastModifier());
+ Assertions.assertEquals("test", alteredTable5.auditInfo().creator());
+ Assertions.assertEquals("test", alteredTable5.auditInfo().lastModifier());
}
@Test
@@ -612,6 +623,40 @@ public class TestTableOperationDispatcher extends
TestOperationDispatcher {
doThrow(new IOException()).when(entityStore).delete(any(), any(),
anyBoolean());
Assertions.assertThrows(
RuntimeException.class, () ->
tableOperationDispatcher.dropTable(tableIdent));
+
+ tableOperationDispatcher.createTable(tableIdent, columns, "comment",
props, new Transform[0]);
+ reset(entityStore);
+ doThrow(new OptimisticLockException("mock conflict"))
+ .when(entityStore)
+ .delete(any(), any(), anyBoolean());
+ Assertions.assertThrows(
+ OptimisticLockException.class, () ->
tableOperationDispatcher.dropTable(tableIdent));
+ }
+
+ @Test
+ public void testPurgeTablePropagatesOptimisticLockConflict() throws
IOException {
+ NameIdentifier tableIdent =
+ NameIdentifier.of(metalake, catalog, "schema_purge_occ",
"table_purge_occ");
+ Map<String, String> props = ImmutableMap.of("k1", "v1", "k2", "v2");
+ Column[] columns =
+ new Column[] {
+ TestColumn.builder()
+ .withName("col1")
+ .withPosition(0)
+ .withType(Types.StringType.get())
+ .build()
+ };
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(tableIdent.namespace().levels()), "comment", props);
+ tableOperationDispatcher.createTable(tableIdent, columns, "comment",
props, new Transform[0]);
+
+ reset(entityStore);
+ doThrow(new OptimisticLockException("mock conflict"))
+ .when(entityStore)
+ .delete(any(), any(), anyBoolean());
+
+ Assertions.assertThrows(
+ OptimisticLockException.class, () ->
tableOperationDispatcher.purgeTable(tableIdent));
}
@Test
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..a743e9108e
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestTableMetaBaseSQLProvider.java
@@ -0,0 +1,66 @@
+/*
+ * 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 testOverwriteAdvancesStoredVersion() {
+ String sql = PROVIDER.insertTableMetaOnDuplicateKeyUpdate(null);
+ String updateClause = sql.substring(sql.indexOf(" ON DUPLICATE KEY
UPDATE"));
+
+ Assertions.assertTrue(updateClause.contains("last_version =
current_version + 1"));
+ Assertions.assertTrue(updateClause.contains("current_version =
current_version + 1"));
+ Assertions.assertFalse(updateClause.contains("current_version =
#{tableMeta.currentVersion}"));
+ Assertions.assertFalse(updateClause.contains("last_version =
#{tableMeta.lastVersion}"));
+ }
+
+ @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..f6c3492a7b
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestTableMetaPostgreSQLProvider.java
@@ -0,0 +1,61 @@
+/*
+ * 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.apache.gravitino.storage.relational.mapper.TableMetaMapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestTableMetaPostgreSQLProvider {
+
+ @Test
+ void testOverwriteAdvancesStoredVersion() {
+ String updateClause = conflictClause();
+
+ // An overwrite must never write the initial version back, or a stale
writer could still pass
+ // its own version check afterwards.
+ Assertions.assertTrue(
+ updateClause.contains(
+ "current_version = " + TableMetaMapper.TABLE_NAME +
".current_version + 1"),
+ () -> "current_version must advance in: " + updateClause);
+ Assertions.assertTrue(
+ updateClause.contains(
+ "last_version = " + TableMetaMapper.TABLE_NAME + ".current_version
+ 1"),
+ () -> "last_version must advance in: " + updateClause);
+
+ // PostgreSQL rejects a bare column name on this side of ON CONFLICT,
because it could mean
+ // either the stored row or the rejected one. Both assignments must name
the table.
+ Assertions.assertFalse(
+ updateClause.matches(".*[^.\\w]current_version\\s*\\+.*"),
+ () -> "Found an unqualified current_version reference in: " +
updateClause);
+ }
+
+ @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"));
+ }
+
+ private String conflictClause() {
+ String sql = new
TableMetaPostgreSQLProvider().insertTableMetaOnDuplicateKeyUpdate(null);
+ return sql.substring(sql.indexOf(" ON CONFLICT"));
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestTablePO.java
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestTablePO.java
new file mode 100644
index 0000000000..45db0357e1
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestTablePO.java
@@ -0,0 +1,88 @@
+/*
+ * 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.po;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestTablePO {
+
+ @Test
+ void testCopyBuilderCarriesEveryField() throws Exception {
+ TablePO source = fullyPopulated();
+
+ TablePO copy = TablePO.builder(source).build();
+
+ // The copy builder lists the fields by hand, so a field added to TablePO
without a matching
+ // line there would be silently blanked on every path that copies a row.
Compare reflectively
+ // instead of naming the fields again here: this fails when the two drift
apart.
+ List<String> dropped = new ArrayList<>();
+ for (Field field : TablePO.class.getDeclaredFields()) {
+ if (Modifier.isStatic(field.getModifiers())) {
+ continue;
+ }
+ field.setAccessible(true);
+ if (!java.util.Objects.equals(field.get(source), field.get(copy))) {
+ dropped.add(field.getName());
+ }
+ }
+ Assertions.assertTrue(
+ dropped.isEmpty(), () -> "TablePO.builder(TablePO) does not copy these
fields: " + dropped);
+ }
+
+ /** Every field set to a distinct value, so a missed copy cannot pass by
coincidence. */
+ private TablePO fullyPopulated() throws Exception {
+ TablePO.Builder builder =
+ TablePO.builder()
+ .withTableId(1L)
+ .withTableName("table")
+ .withMetalakeId(2L)
+ .withCatalogId(3L)
+ .withSchemaId(4L)
+ .withAuditInfo("audit")
+ .withCurrentVersion(5L)
+ .withLastVersion(6L)
+ .withDeletedAt(7L)
+ .withFormat("format")
+ .withProperties("properties")
+ .withPartitions("partitions")
+ .withSortOrders("sortOrders")
+ .withDistribution("distribution")
+ .withIndexes("indexes")
+ .withComment("comment");
+ TablePO po = builder.build();
+
+ // Guard the guard: if a new field is added and left unset above, the
comparison would trivially
+ // hold with null on both sides and prove nothing.
+ for (Field field : TablePO.class.getDeclaredFields()) {
+ if (Modifier.isStatic(field.getModifiers())) {
+ continue;
+ }
+ field.setAccessible(true);
+ Assertions.assertNotNull(
+ field.get(po),
+ "Set TablePO." + field.getName() + " in this test so the copy check
covers it");
+ }
+ return po;
+ }
+}
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 40d737bca9..2ff71674de 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
@@ -40,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;
@@ -58,19 +59,24 @@ import
org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.expressions.transforms.Transforms;
import org.apache.gravitino.rel.indexes.Index;
import org.apache.gravitino.rel.indexes.Indexes;
+import org.apache.gravitino.rel.types.Type;
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;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.function.Executable;
public class TestTableMetaService extends TestJDBCBackend {
private final String metalakeName = "metalake_for_table_test";
@@ -109,6 +115,198 @@ 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);
+
+ Throwable insertFailure =
+ runWhileSchemaDeleteUncommitted(
+ observedSchemaPO, () ->
TableMetaService.getInstance().insertTable(table, false));
+
+ Assertions.assertInstanceOf(NoSuchEntityException.class, insertFailure);
+ Assertions.assertTrue(
+ SessionUtils.getWithoutCommit(
+ TableMetaMapper.class, mapper ->
mapper.listTablePOsByTableIds(List.of(table.id())))
+ .isEmpty());
+ }
+
+ @TestTemplate
+ public void testInsertRollsBackAllRowsWhenColumnWriteFails() throws
IOException {
+ createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO);
+ Namespace tableNamespace = NamespaceUtil.ofTable(metalakeName,
catalogName, schemaName);
+ ColumnEntity column = column("column", Types.IntegerType.get());
+ 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 = column("original_column",
Types.IntegerType.get());
+ 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 = column("replacement_column",
Types.StringType.get());
+ 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 testOverwriteAdvancesVersionAndRejectsStaleUpdate() throws
IOException {
+ createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO);
+ TableEntity original =
+ createTableEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofTable(metalakeName, catalogName, schemaName),
+ "table_overwrite_occ",
+ AUDIT_INFO);
+ TableMetaService.getInstance().insertTable(original, false);
+ TablePO beforeOverwrite = getTablePO(original.id());
+ TableEntity replacement = copyTableWithComment(original, "overwrite
winner");
+
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ updateTableUnchecked(
+ original.nameIdentifier(),
+ current -> {
+ insertTableUnchecked(replacement, true);
+ return copyTableWithComment(current, "stale update");
+ }));
+
+ TableEntity winner =
+
TableMetaService.getInstance().getTableByIdentifier(original.nameIdentifier());
+ TablePO afterOverwrite = getTablePO(original.id());
+ Assertions.assertEquals("overwrite winner", winner.comment());
+ Assertions.assertEquals(
+ beforeOverwrite.getCurrentVersion() + 1,
afterOverwrite.getCurrentVersion());
+ Assertions.assertEquals(afterOverwrite.getCurrentVersion(),
afterOverwrite.getLastVersion());
+ }
+
+ @TestTemplate
+ public void testNaturalKeyOverwriteUsesPersistedTableId() throws IOException
{
+ // PostgreSQL's upsert targets table_id and rejects a different ID on the
natural key before
+ // readback. This regression covers MySQL/H2 ON DUPLICATE KEY, which can
choose either key.
+ Assumptions.assumeFalse("postgresql".equalsIgnoreCase(backendType));
+ createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO);
+ Namespace tableNamespace = NamespaceUtil.ofTable(metalakeName,
catalogName, schemaName);
+ TableEntity original =
+ TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("table_natural_key_overwrite")
+ .withNamespace(tableNamespace)
+ .withColumns(List.of(column("original_column",
Types.IntegerType.get())))
+ .withComment("original")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(original, false);
+ TablePO beforeOverwrite = getTablePO(original.id());
+ TableEntity replacement =
+ TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(original.name())
+ .withNamespace(tableNamespace)
+ .withColumns(List.of(column("replacement_column",
Types.StringType.get())))
+ .withComment("replacement")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+
+ TableMetaService.getInstance().insertTable(replacement, true);
+
+ TableEntity stored =
+
TableMetaService.getInstance().getTableByIdentifier(original.nameIdentifier());
+ TablePO afterOverwrite = getTablePO(original.id());
+ Assertions.assertEquals(original.id(), stored.id());
+ Assertions.assertEquals("replacement", stored.comment());
+ Assertions.assertEquals("replacement_column",
stored.columns().get(0).name());
+ Assertions.assertEquals(
+ beforeOverwrite.getCurrentVersion() + 1,
afterOverwrite.getCurrentVersion());
+ }
+
@TestTemplate
public void testUpdateAlreadyExistsException() throws IOException {
createAndInsertMakeLake(metalakeName);
@@ -326,6 +524,208 @@ 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");
+
+ assertStaleAlterReportsNoSuch(
+ table,
+ competing ->
+ copyTable(competing, renamedIdentifier.name(),
competing.namespace(), "renamed winner"),
+ renamedIdentifier,
+ "renamed winner");
+ }
+
+ @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());
+
+ assertStaleAlterReportsNoSuch(
+ table,
+ competing -> copyTable(competing, competing.name(), movedNamespace,
"moved winner"),
+ movedIdentifier,
+ "moved winner");
+ }
+
+ @TestTemplate
+ public void testDeleteRejectsStaleVersion() throws IOException {
+ createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO);
+ ColumnEntity column = column("column_that_must_survive",
Types.IntegerType.get());
+ 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 = column("duplicate_id",
Types.IntegerType.get());
+ // 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 copyTableWithColumns(
+ current, List.of(duplicateColumn, duplicateColumn),
"must roll back");
+ }));
+
+ 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 testMoveTableWaitsForConcurrentTargetSchemaDelete() throws
Exception {
String sourceSchemaName = "source_schema";
@@ -349,73 +749,24 @@ public class TestTableMetaService extends TestJDBCBackend
{
SchemaPO observedTargetSchema =
SessionUtils.getWithoutCommit(
SchemaMetaMapper.class, mapper ->
mapper.selectSchemaMetaById(targetSchema.id()));
- CountDownLatch targetDeleteLocked = new CountDownLatch(1);
- CountDownLatch allowDeleteCommit = new CountDownLatch(1);
- CountDownLatch moveStarted = 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(
- observedTargetSchema.getSchemaId(),
-
observedTargetSchema.getCurrentVersion()));
- Assertions.assertEquals(1, deleted);
- targetDeleteLocked.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(targetDeleteLocked.await(30, TimeUnit.SECONDS));
- Future<Throwable> moveResult =
- executor.submit(
- () -> {
- moveStarted.countDown();
- try {
- TableEntity movedTable =
- TableEntity.builder()
- .withId(table.id())
- .withName(table.name())
- .withNamespace(
- NamespaceUtil.ofTable(metalakeName, catalogName,
targetSchemaName))
- .withColumns(table.columns())
- .withAuditInfo(table.auditInfo())
- .build();
- backend.update(
- table.nameIdentifier(), Entity.EntityType.TABLE, ignored
-> movedTable);
- return null;
- } catch (Throwable throwable) {
- return throwable;
- }
- });
- assertTrue(moveStarted.await(30, TimeUnit.SECONDS));
- // Resolving the target ID happens before the table transaction. The
move must then wait on
- // the target schema row instead of writing below a schema whose delete
is about to commit.
- assertThrows(TimeoutException.class, () -> moveResult.get(500,
TimeUnit.MILLISECONDS));
+ TableEntity movedTable =
+ TableEntity.builder()
+ .withId(table.id())
+ .withName(table.name())
+ .withNamespace(NamespaceUtil.ofTable(metalakeName, catalogName,
targetSchemaName))
+ .withColumns(table.columns())
+ .withAuditInfo(table.auditInfo())
+ .build();
+ // Resolving the target ID happens before the table transaction. The move
must then wait on the
+ // target schema row instead of writing below a schema whose delete is
about to commit.
+ Throwable moveFailure =
+ runWhileSchemaDeleteUncommitted(
+ observedTargetSchema,
+ () ->
+ backend.update(
+ table.nameIdentifier(), Entity.EntityType.TABLE, ignored
-> movedTable));
- allowDeleteCommit.countDown();
- Assertions.assertNull(deleteResult.get(30, TimeUnit.SECONDS));
- Assertions.assertInstanceOf(
- NoSuchEntityException.class, moveResult.get(30, TimeUnit.SECONDS));
- } finally {
- allowDeleteCommit.countDown();
- executor.shutdownNow();
- }
+ Assertions.assertInstanceOf(NoSuchEntityException.class, moveFailure);
TableEntity unchanged =
TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier());
@@ -498,16 +849,7 @@ public class TestTableMetaService extends TestJDBCBackend {
createAndInsertCatalog(metalakeName, catalogName);
createAndInsertSchema(metalakeName, catalogName, schemaName);
- ColumnEntity column =
- ColumnEntity.builder()
- .withId(RandomIdGenerator.INSTANCE.nextId())
- .withName("col1")
- .withPosition(0)
- .withDataType(Types.IntegerType.get())
- .withNullable(true)
- .withAutoIncrement(false)
- .withAuditInfo(AUDIT_INFO)
- .build();
+ ColumnEntity column = column("col1", Types.IntegerType.get());
TableEntity table =
TableEntity.builder()
.withId(RandomIdGenerator.INSTANCE.nextId())
@@ -557,4 +899,167 @@ 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));
+ }
+
+ /**
+ * Holds an uncommitted delete of the given schema open, runs {@code victim}
against it, and
+ * returns what the victim threw once the delete commits, or null if it
succeeded.
+ *
+ * <p>The helper also asserts the part both callers care about: while the
delete is in flight the
+ * victim must block on the schema row rather than slip past it.
+ */
+ private Throwable runWhileSchemaDeleteUncommitted(SchemaPO observedSchemaPO,
Executable victim)
+ throws Exception {
+ CountDownLatch schemaDeleteLocked = new CountDownLatch(1);
+ CountDownLatch allowDeleteCommit = new CountDownLatch(1);
+ CountDownLatch victimStarted = 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> victimResult =
+ executor.submit(
+ () -> {
+ victimStarted.countDown();
+ try {
+ victim.execute();
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ assertTrue(victimStarted.await(30, TimeUnit.SECONDS));
+ assertThrows(TimeoutException.class, () -> victimResult.get(500,
TimeUnit.MILLISECONDS));
+
+ allowDeleteCommit.countDown();
+ Assertions.assertNull(deleteResult.get(30, TimeUnit.SECONDS));
+ return victimResult.get(30, TimeUnit.SECONDS);
+ } finally {
+ allowDeleteCommit.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ /**
+ * Runs an alter that loses to a competing writer which renames or moves the
table away, and
+ * asserts the loser is told the table is gone while the winner's change
survives.
+ */
+ private void assertStaleAlterReportsNoSuch(
+ TableEntity table,
+ Function<TableEntity, TableEntity> competingUpdate,
+ NameIdentifier winnerIdentifier,
+ String winnerComment) {
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ TableMetaService.getInstance()
+ .updateTable(
+ table.nameIdentifier(),
+ entity -> {
+ TableEntity current = (TableEntity) entity;
+ updateTableUnchecked(table.nameIdentifier(),
competingUpdate);
+ return copyTableWithComment(current, "stale update");
+ }));
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier()));
+ Assertions.assertEquals(
+ winnerComment,
+
TableMetaService.getInstance().getTableByIdentifier(winnerIdentifier).comment());
+ }
+
+ private ColumnEntity column(String name, Type dataType) {
+ return ColumnEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(name)
+ .withPosition(0)
+ .withDataType(dataType)
+ .withNullable(true)
+ .withAutoIncrement(false)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
+ private TableEntity copyTableWithComment(TableEntity current, String
comment) {
+ return copyTable(current, current.name(), current.namespace(), comment);
+ }
+
+ private TableEntity copyTableWithColumns(
+ TableEntity current, List<ColumnEntity> columns, String comment) {
+ return copyTable(current, current.name(), current.namespace(), comment,
columns);
+ }
+
+ private TableEntity copyTable(
+ TableEntity current, String name, Namespace namespace, String comment) {
+ return copyTable(current, name, namespace, comment, current.columns());
+ }
+
+ private TableEntity copyTable(
+ TableEntity current,
+ String name,
+ Namespace namespace,
+ String comment,
+ List<ColumnEntity> columns) {
+ return TableEntity.builder()
+ .withId(current.id())
+ .withName(name)
+ .withNamespace(namespace)
+ .withColumns(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);
+ }
+ }
+
+ private void insertTableUnchecked(TableEntity table, boolean overwrite) {
+ try {
+ TableMetaService.getInstance().insertTable(table, overwrite);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
}
