This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 8433c5671c [#13949]fix(core): Preserve table metadata on load repair
conflicts (#13930)
8433c5671c is described below
commit 8433c5671c9f4393f9699869599b4cd41eab14ba
Author: Qi Yu <[email protected]>
AuthorDate: Tue Oct 20 21:38:27 2026 +0800
[#13949]fix(core): Preserve table metadata on load repair conflicts (#13930)
### What changes were proposed in this pull request?
- Skip table-version writes when the existing conditional table-meta
update affects no rows, matching the existing guard for column writes.
- During load-time column repair only, recognize the existing
table-update conflict message, log a warning, and retain the metadata
snapshot already read by that load. Other failures keep their existing
handling.
### Why are the changes needed?
A concurrent metadata update can win after a table load reads its
snapshot. On branch-1.3, the losing update still writes the version
payload before reporting a conflict, potentially overwriting the winner.
The load then drops its Gravitino metadata because the best-effort
repair failed.
This is a limited branch-1.3 fix. It preserves the existing
conditional-update mechanism without backporting the OCC framework or
replaying external DDL. The returned metadata snapshot can temporarily
be stale; broader HA read consistency and stale catalog snapshots remain
outside this patch.
Fixed: #13949
### Does this PR introduce _any_ user-facing change?
Table loads retain their already-read Gravitino metadata when column
repair loses a concurrent update. No API, configuration, or schema
changes.
### How was this patch tested?
- Focused `TestTableMetaService` and `TestTableOperationDispatcher`
suites: 34 passed.
- Full core non-Docker unit tests: 1,656 passed, 2 skipped on the final
rerun.
- `TestTableMetaService` with H2, MySQL, and PostgreSQL backends: all 21
passed.
- Mutation verification: reverting the production fixes makes both new
regression tests fail (lost load metadata and overwritten winning
version).
- Core Spotless checks.
An initial full-core run hit three metrics initialization failures. A
separate same-JVM probe using unmodified branch-1.3 service code/tests
reproduced the existing fixture ordering dependency (JDBC tests leave a
mock global Config without the metrics parameter). The final full-core
rerun passed.
---
.../catalog/TableOperationDispatcher.java | 24 +++++++-
.../relational/service/TableMetaService.java | 9 ++-
.../catalog/TestTableOperationDispatcher.java | 51 +++++++++++++++
.../relational/service/TestTableMetaService.java | 72 ++++++++++++++++++++++
4 files changed, 151 insertions(+), 5 deletions(-)
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 43cb76ff7c..725a688467 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
@@ -23,6 +23,7 @@ import static org.apache.gravitino.Entity.EntityType.TABLE;
import static org.apache.gravitino.catalog.CapabilityHelpers.applyCapabilities;
import static
org.apache.gravitino.catalog.PropertiesMetadataHelpers.validatePropertyForCreate;
import static
org.apache.gravitino.rel.expressions.transforms.Transforms.EMPTY_TRANSFORM;
+import static
org.apache.gravitino.storage.relational.service.TableMetaService.UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX;
import static
org.apache.gravitino.utils.NameIdentifierUtil.getCatalogIdentifier;
import static
org.apache.gravitino.utils.NameIdentifierUtil.getSchemaIdentifier;
@@ -30,6 +31,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
+import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
@@ -984,8 +986,9 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
() ->
operateOnEntity(
tableIdent,
- id ->
- store.update(
+ id -> {
+ try {
+ return store.update(
id,
TableEntity.class,
TABLE,
@@ -1012,7 +1015,22 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
PrincipalUtils.getCurrentPrincipal().getName())
.withLastModifiedTime(Instant.now())
.build())
- .build()),
+ .build());
+ } catch (IOException e) {
+ if (!(UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX +
id).equals(e.getMessage())) {
+ throw e;
+ }
+ // Column repair is best-effort. Retain this load's
existing snapshot rather
+ // than lose its metadata or replay a write against a
concurrent change.
+ // This snapshot can be stale; this does not provide HA
read consistency.
+ LOG.warn(
+ "Skipping column repair for table {} after a
concurrent metadata update; "
+ + "retaining the metadata snapshot read by this
load",
+ id,
+ e);
+ return combinedTable.tableFromGravitino();
+ }
+ },
"UPDATE",
combinedTable.tableFromGravitino().id()));
}
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 17ee0d5c3b..5a01a10015 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
@@ -210,14 +210,19 @@ public class TableMetaService {
SessionUtils.getWithoutCommit(
TableMetaMapper.class,
mapper -> ops.updatePO(mapper, newTablePO, oldTablePO))),
- () ->
+ () -> {
+ // A concurrent update may have won. Its version payload must not
be overwritten by
+ // this stale update when the conditional table-meta update
affected no rows.
+ if (updateResult.get() > 0) {
SessionUtils.doWithoutCommit(
TableVersionMapper.class,
mapper -> {
mapper.softDeleteTableVersionByTableIdAndVersion(
oldTablePO.getTableId(),
oldTablePO.getCurrentVersion());
mapper.insertTableVersionOnDuplicateKeyUpdate(newTablePO);
- }),
+ });
+ }
+ },
() -> {
if (updateResult.get() > 0) {
TableColumnMetaService.getInstance()
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 0a819235bc..598163c506 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
@@ -33,6 +33,8 @@ import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import com.google.common.collect.ImmutableMap;
import java.io.IOException;
@@ -75,6 +77,7 @@ import org.apache.gravitino.rel.TableChange;
import org.apache.gravitino.rel.expressions.literals.Literals;
import org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.types.Types;
+import org.apache.gravitino.storage.relational.service.TableMetaService;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -1060,6 +1063,54 @@ public class TestTableOperationDispatcher extends
TestOperationDispatcher {
Assertions.assertEquals(ids.get("b"), afterDropAndAdd.get("c"));
}
+ @Test
+ void testLoadRetainsSnapshotOnlyForTableUpdateConflict() throws IOException {
+ reset(entityStore);
+ Namespace namespace = Namespace.of(metalake, catalog,
"schema_load_conflict");
+ NameIdentifier ident = NameIdentifier.of(namespace, "table_load_conflict");
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(namespace.levels()), "comment",
ImmutableMap.of("k1", "v1", "k2", "v2"));
+ Column[] columns = {
+
TestColumn.builder().withName("c1").withPosition(0).withType(Types.IntegerType.get()).build()
+ };
+ tableOperationDispatcher.createTable(
+ ident, columns, "comment", ImmutableMap.of("k1", "v1", "k2", "v2"),
new Transform[0]);
+ TableEntity snapshot = entityStore.get(ident, TABLE, TableEntity.class);
+ TestCatalog testCatalog =
+ (TestCatalog)
+ catalogManager.loadCatalogAndWrap(NameIdentifier.of(metalake,
catalog)).catalog();
+ ((TestCatalogOperations) testCatalog.ops())
+ .alterTable(ident, TableChange.renameColumn(new String[] {"c1"},
"c1_new"));
+ try {
+ doThrow(new
IOException(TableMetaService.UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX + ident))
+ .when(entityStore)
+ .update(any(), any(), any(), any());
+ EntityCombinedTable loaded = (EntityCombinedTable)
tableOperationDispatcher.loadTable(ident);
+ Assertions.assertEquals(snapshot, loaded.tableFromGravitino());
+ Assertions.assertEquals("c1_new", loaded.columns()[0].name());
+ verify(entityStore, times(1)).update(any(), any(), any(), any());
+ Assertions.assertEquals(snapshot, entityStore.get(ident, TABLE,
TableEntity.class));
+
+ reset(entityStore);
+ doThrow(new IOException("Storage unavailable"))
+ .when(entityStore)
+ .update(any(), any(), any(), any());
+ Assertions.assertNull(
+ ((EntityCombinedTable)
tableOperationDispatcher.loadTable(ident)).tableFromGravitino());
+
+ reset(entityStore);
+ doThrow(
+ new IOException(
+ TableMetaService.UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX +
"another_table"))
+ .when(entityStore)
+ .update(any(), any(), any(), any());
+ Assertions.assertNull(
+ ((EntityCombinedTable)
tableOperationDispatcher.loadTable(ident)).tableFromGravitino());
+ } finally {
+ reset(entityStore);
+ }
+ }
+
@Test
public void testLoadDoesNotUndoConcurrentColumnRename() throws IOException {
Namespace tableNs = Namespace.of(metalake, catalog, "schema_load_race");
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 08d874f295..5399f3c0f9 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
@@ -177,6 +177,78 @@ public class TestTableMetaService extends TestJDBCBackend {
assertFalse(legacyRecordExistsInDB(table.id(), Entity.EntityType.TABLE));
}
+ @TestTemplate
+ void testConflictingUpdatePreservesWinningVersion() throws IOException {
+ createParentEntities(metalakeName, catalogName, schemaName, AUDIT_INFO);
+ Namespace namespace = Namespace.of(metalakeName, catalogName, schemaName);
+ long tableId = RandomIdGenerator.INSTANCE.nextId();
+ ColumnEntity column =
+ ColumnEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("column1")
+ .withPosition(0)
+ .withDataType(Types.IntegerType.get())
+ .withNullable(true)
+ .withAutoIncrement(false)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableEntity original =
+ TableEntity.builder()
+ .withId(tableId)
+ .withName("conflicting_table")
+ .withNamespace(namespace)
+ .withColumns(List.of(column))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService service = TableMetaService.getInstance();
+ service.insertTable(original, false);
+ TableEntity winner =
+ TableEntity.builder()
+ .withId(tableId)
+ .withName(original.name())
+ .withNamespace(namespace)
+ .withComment("winner")
+ .withProperties(ImmutableMap.of("key", "winner"))
+ .withColumns(List.of(column))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableEntity loser =
+ TableEntity.builder()
+ .withId(tableId)
+ .withName(original.name())
+ .withNamespace(namespace)
+ .withComment("loser")
+ .withProperties(ImmutableMap.of("key", "loser"))
+ .withColumns(List.of())
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+
+ IOException conflict =
+ assertThrows(
+ IOException.class,
+ () ->
+ service.updateTable(
+ original.nameIdentifier(),
+ (TableEntity old) -> {
+ // The outer update has read its old snapshot, but has
not begun its write
+ // transaction.
+ try {
+ service.updateTable(
+ original.nameIdentifier(), (TableEntity ignored)
-> winner);
+ } catch (IOException e) {
+ throw new AssertionError(e);
+ }
+ return loser;
+ }));
+ Assertions.assertEquals(
+ TableMetaService.UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX +
original.nameIdentifier(),
+ conflict.getMessage());
+ TableEntity stored =
service.getTableByIdentifier(original.nameIdentifier());
+ Assertions.assertEquals(winner.comment(), stored.comment());
+ Assertions.assertEquals(winner.properties(), stored.properties());
+ compareTwoColumns(winner.columns(), stored.columns());
+ }
+
@TestTemplate
public void testUpdateTable() throws IOException {
String catalogName = "catalog1";
