From a0510b46c023fb5556f36e6aff46aa01157f1bae Mon Sep 17 00:00:00 2001 From: yuqi Date: Fri, 3 Jul 2026 16:49:58 +0800 Subject: [PATCH 1/5] [#11891] fix(lance): retry repair-on-load metadata update on optimistic-lock conflict repairTableMetadata and recordCheckedEmptyVersion run an optimistic-locked EntityStore.update on every loadTable. Concurrent loads repairing the same table race on the version CAS; the loser gets IOException("Failed to update the entity"), which was rethrown as a fatal RuntimeException (HTTP 500). Wrap the update in a bounded CAS retry (updateTableWithCasRetry): on conflict, re-read the latest already-repaired entity and re-apply the idempotent updater, returning a usable table instead of failing the load. Fix: #11891 Signed-off-by: yuqi --- .../lakehouse/lance/LanceTableOperations.java | 41 +++++++-- .../lance/TestLanceTableOperations.java | 86 +++++++++++++++++++ 2 files changed, 121 insertions(+), 6 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 38dbf0e90d2..8aefae6a48c 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 @@ -31,6 +31,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.IntStream; import org.apache.arrow.vector.types.pojo.Field; @@ -79,6 +80,12 @@ public class LanceTableOperations extends ManagedTableOperations { private static final Logger LOG = LoggerFactory.getLogger(LanceTableOperations.class); + /** + * Max attempts for the optimistic-locked repair-on-load {@code store.update}. Concurrent loads + * can repair the same table at once; the loser of the version CAS re-reads and retries. + */ + private static final int REPAIR_UPDATE_MAX_ATTEMPTS = 5; + public enum CreationMode { CREATE, EXIST_OK, @@ -527,10 +534,8 @@ private Column[] extractColumns(Schema arrowSchema) { private Table repairTableMetadata(NameIdentifier ident, Column[] columns, long datasetVersion) { try { TableEntity tableEntity = - store.update( + updateTableWithCasRetry( ident, - TableEntity.class, - Entity.EntityType.TABLE, current -> { if (!needsSchemaRefresh(current, datasetVersion)) { return current; @@ -547,13 +552,37 @@ private Table repairTableMetadata(NameIdentifier ident, Column[] columns, long d } } + /** + * 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}. Because the updater is idempotent, the loser re-reads the latest + * (already repaired) entity and retries instead of failing the whole load with a fatal error. + */ + private TableEntity updateTableWithCasRetry( + NameIdentifier ident, Function updater) throws IOException { + IOException 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) { + lastConflict = e; + LOG.debug( + "Optimistic-lock conflict updating table {} metadata (attempt {}/{}), retrying", + ident, + attempt, + REPAIR_UPDATE_MAX_ATTEMPTS, + e); + } + } + throw lastConflict; + } + private Table recordCheckedEmptyVersion(NameIdentifier ident, long datasetVersion) { try { TableEntity tableEntity = - store.update( + updateTableWithCasRetry( ident, - TableEntity.class, - Entity.EntityType.TABLE, current -> { if (!isDatasetVersionChanged(current.properties(), datasetVersion)) { return current; 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 59b2a41ff9e..478b525e74c 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 @@ -34,6 +34,7 @@ import static org.mockito.Mockito.when; import com.google.common.collect.Maps; +import java.io.IOException; import java.time.Instant; import java.util.List; import java.util.Map; @@ -170,6 +171,91 @@ public void testLoadDeclaredTableSchemaFromLocation() throws Exception { Assertions.assertFalse(loadedTable.properties().containsKey(LANCE_TABLE_DECLARED)); } + /** + * 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")}. {@code repairTableMetadata} currently rethrows it as a fatal {@code + * RuntimeException} (HTTP 500) instead of tolerating the concurrent update. A correct fix should + * treat the lost race as benign and return a usable table, so this test asserts the desired + * behavior and currently fails against the buggy code. + */ + @Test + public void testLoadTableSurvivesConcurrentRepairVersionRace() throws Exception { + NameIdentifier ident = NameIdentifier.of("schema", "table"); + String location = tempDir.resolve("concurrent-repair-table").toString(); + TableEntity tableEntity = + tableEntity( + ident, + List.of(), + Map.of( + Table.PROPERTY_LOCATION, + location, + LANCE_TABLE_DECLARED, + "true", + LANCE_STORAGE_OPTIONS_PREFIX + "endpoint", + "http://endpoint")); + // The winner of the race already repaired the table to the dataset schema and version. + TableEntity alreadyRepairedTableEntity = + tableEntity( + ident, + List.of( + ColumnEntity.builder() + .withId(11L) + .withName("id") + .withDataType(Types.IntegerType.get()) + .withPosition(0) + .withAuditInfo(AuditInfo.EMPTY) + .build(), + ColumnEntity.builder() + .withId(12L) + .withName("name") + .withDataType(Types.StringType.get()) + .withPosition(1) + .withAuditInfo(AuditInfo.EMPTY) + .build()), + Map.of(Table.PROPERTY_LOCATION, location, LANCE_TABLE_VERSION, "8")); + when(store.get(eq(ident), eq(Entity.EntityType.TABLE), eq(TableEntity.class))) + .thenReturn(tableEntity); + 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. + when(store.update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE), any())) + .thenThrow(new IOException("Failed to update the entity: " + ident)) + .thenAnswer( + invocation -> { + @SuppressWarnings("unchecked") + Function updater = invocation.getArgument(3); + return updater.apply(alreadyRepairedTableEntity); + }); + + Dataset dataset = mock(Dataset.class); + when(dataset.getSchema()) + .thenReturn( + new Schema( + List.of( + Field.nullable("id", new ArrowType.Int(32, true)), + Field.nullable("name", new ArrowType.Utf8())))); + when(dataset.version()).thenReturn(8L); + Mockito.doReturn(dataset) + .when(lanceTableOps) + .openDataset(location, Map.of("endpoint", "http://endpoint")); + + // A lost repair race must not fail the load: the bounded CAS retry recovers and returns the + // repaired table. Fails today because repairTableMetadata rethrows the first conflict as + // RuntimeException("Failed to repair table"). + Table loadedTable = + Assertions.assertDoesNotThrow( + () -> + PrincipalUtils.doAs( + new UserPrincipal("tester"), () -> lanceTableOps.loadTable(ident))); + Assertions.assertEquals(2, loadedTable.columns().length); + Assertions.assertEquals("id", loadedTable.columns()[0].name()); + Assertions.assertEquals("name", loadedTable.columns()[1].name()); + } + @Test public void testLoadTableWithStoredColumnsDoesNotReadLocation() throws Exception { NameIdentifier ident = NameIdentifier.of("schema", "table"); From d79c7b2963828172c8de59ae9776c472ec26c691 Mon Sep 17 00:00:00 2001 From: yuqi Date: Fri, 3 Jul 2026 19:19:04 +0800 Subject: [PATCH 2/5] [#11891] fix(lance): address repair retry review comments --- .../catalog/lakehouse/lance/LanceTableOperations.java | 3 ++- .../lakehouse/lance/TestLanceTableOperations.java | 11 +++++------ 2 files changed, 7 insertions(+), 7 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 8aefae6a48c..bb7759d643f 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 @@ -568,10 +568,11 @@ private TableEntity updateTableWithCasRetry( } catch (IOException e) { lastConflict = e; LOG.debug( - "Optimistic-lock conflict updating table {} metadata (attempt {}/{}), retrying", + "Optimistic-lock conflict updating table {} metadata (attempt {}/{}), {}", ident, attempt, REPAIR_UPDATE_MAX_ATTEMPTS, + attempt < REPAIR_UPDATE_MAX_ATTEMPTS ? "retrying" : "retries exhausted", e); } } 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 478b525e74c..bf99a64814b 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 @@ -175,10 +175,9 @@ public void testLoadDeclaredTableSchemaFromLocation() throws Exception { * 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")}. {@code repairTableMetadata} currently rethrows it as a fatal {@code - * RuntimeException} (HTTP 500) instead of tolerating the concurrent update. A correct fix should - * treat the lost race as benign and return a usable table, so this test asserts the desired - * behavior and currently fails against the buggy code. + * 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. */ @Test public void testLoadTableSurvivesConcurrentRepairVersionRace() throws Exception { @@ -244,8 +243,8 @@ public void testLoadTableSurvivesConcurrentRepairVersionRace() throws Exception .openDataset(location, Map.of("endpoint", "http://endpoint")); // A lost repair race must not fail the load: the bounded CAS retry recovers and returns the - // repaired table. Fails today because repairTableMetadata rethrows the first conflict as - // RuntimeException("Failed to repair table"). + // repaired table instead of surfacing the first conflict as RuntimeException("Failed to repair + // table"). Table loadedTable = Assertions.assertDoesNotThrow( () -> From b41e0d02e09be8828e094b5c0fed0ed02a8c0b59 Mon Sep 17 00:00:00 2001 From: Qi Yu Date: Fri, 3 Jul 2026 20:11:14 +0800 Subject: [PATCH 3/5] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../catalog/lakehouse/lance/LanceTableOperations.java | 7 +++++++ 1 file changed, 7 insertions(+) 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 bb7759d643f..4aab2d6b29e 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 @@ -566,6 +566,13 @@ private TableEntity updateTableWithCasRetry( 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("Failed to update the entity:")) { + throw e; + } + lastConflict = e; LOG.debug( "Optimistic-lock conflict updating table {} metadata (attempt {}/{}), {}", From 27be426c2624fe23c77660d113ff3aed013218e6 Mon Sep 17 00:00:00 2001 From: yuqi Date: Mon, 6 Jul 2026 14:35:52 +0800 Subject: [PATCH 4/5] [#11891] test(lance): add real multi-threaded repair-on-load CAS race test The existing testLoadTableSurvivesConcurrentRepairVersionRace is mock-based and deterministic (one scripted IOException, no threads). Add a genuine multi-threaded stress test that drives loadTable from several threads at once against a CasEntityStore modelling the relational store's optimistic-lock CAS (UPDATE ... WHERE current_version = old, version bumped on every update). Verified: fails at iteration 0 against the pre-fix code with RuntimeException("Failed to repair table"), passes with the bounded CAS retry. Signed-off-by: yuqi --- .../TestLanceConcurrentRepairStress.java | 276 ++++++++++++++++++ 1 file changed, 276 insertions(+) create mode 100644 catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceConcurrentRepairStress.java 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 new file mode 100644 index 00000000000..e2fa8a15931 --- /dev/null +++ b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceConcurrentRepairStress.java @@ -0,0 +1,276 @@ +/* + * 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.catalog.lakehouse.lance; + +import static org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_STORAGE_OPTIONS_PREFIX; +import static org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_TABLE_DECLARED; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.Schema; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.gravitino.Config; +import org.apache.gravitino.Entity; +import org.apache.gravitino.Entity.EntityType; +import org.apache.gravitino.EntityStore; +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.meta.AuditInfo; +import org.apache.gravitino.meta.TableEntity; +import org.apache.gravitino.rel.Table; +import org.apache.gravitino.storage.IdGenerator; +import org.apache.gravitino.utils.Executable; +import org.apache.gravitino.utils.PrincipalUtils; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.lance.Dataset; +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 + * 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}). + * + *

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. + */ +public class TestLanceConcurrentRepairStress { + + @TempDir private java.nio.file.Path tempDir; + + private static final NameIdentifier IDENT = NameIdentifier.of("schema", "table"); + + @Test + public void testConcurrentRepairLoadsSurviveCasContention() throws Exception { + // The reported race is "two loads repair the same table at once". Also drive a small herd to + // confirm the bounded CAS retry stays robust beyond the minimal two-load case. + runStress(2, 3000); + runStress(8, 2000); + } + + private void runStress(int concurrency, int iterations) throws Exception { + String location = tempDir.resolve("cas-race-" + concurrency).toString(); + Map storageOptions = Map.of("endpoint", "http://endpoint"); + Schema datasetSchema = + new Schema( + List.of( + Field.nullable("id", new ArrowType.Int(32, true)), + Field.nullable("name", new ArrowType.Utf8()))); + + ManagedSchemaOperations schemaOps = mock(ManagedSchemaOperations.class); + IdGenerator idGenerator = mock(IdGenerator.class); + AtomicLong idSeq = new AtomicLong(1000); + when(idGenerator.nextId()).thenAnswer(invocation -> idSeq.incrementAndGet()); + + ExecutorService pool = Executors.newFixedThreadPool(concurrency); + UserPrincipal user = new UserPrincipal("tester"); + int failures = 0; + try { + for (int iter = 0; iter < iterations; iter++) { + // Fresh stale (declared, empty-schema) entity every iteration so each round starts from the + // pre-repair state that triggers repairTableMetadata on every concurrent load. + TableEntity stale = + tableEntity( + List.of(), + Map.of( + Table.PROPERTY_LOCATION, + location, + LANCE_TABLE_DECLARED, + "true", + LANCE_STORAGE_OPTIONS_PREFIX + "endpoint", + "http://endpoint")); + CasEntityStore store = new CasEntityStore(stale); + LanceTableOperations ops = spy(new LanceTableOperations(store, schemaOps, idGenerator)); + Dataset dataset = mock(Dataset.class); + when(dataset.version()).thenReturn(8L); + when(dataset.getSchema()).thenReturn(datasetSchema); + Mockito.doReturn(dataset).when(ops).openDataset(location, storageOptions); + + CyclicBarrier barrier = new CyclicBarrier(concurrency); + List> futures = new java.util.ArrayList<>(concurrency); + for (int t = 0; t < concurrency; t++) { + Callable task = + () -> { + barrier.await(); + return PrincipalUtils.doAs(user, () -> ops.loadTable(IDENT)); + }; + futures.add(pool.submit(task)); + } + + for (Future
future : futures) { + try { + Table loaded = future.get(30, TimeUnit.SECONDS); + Assertions.assertEquals( + 2, loaded.columns().length, "repaired table must expose both columns"); + } catch (ExecutionException e) { + failures++; + // Surface the first real failure with its cause for debugging. + if (failures == 1) { + throw new AssertionError( + "Concurrent repair-on-load failed at iteration " + + iter + + " with concurrency=" + + concurrency, + e.getCause()); + } + } + } + } + } finally { + pool.shutdownNow(); + } + Assertions.assertEquals( + 0, failures, "no concurrent load should fail (concurrency=" + concurrency + ")"); + } + + private static TableEntity tableEntity( + List columns, Map properties) { + return TableEntity.builder() + .withId(1L) + .withName(IDENT.name()) + .withNamespace(IDENT.namespace()) + .withComment("comment") + .withColumns(columns) + .withProperties(properties) + .withAuditInfo( + AuditInfo.builder().withCreator("creator").withCreateTime(Instant.EPOCH).build()) + .build(); + } + + /** + * 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. + */ + private static final class CasEntityStore implements EntityStore { + + private static final class Versioned { + private final long version; + private final TableEntity entity; + + Versioned(long version, TableEntity entity) { + this.version = version; + this.entity = entity; + } + } + + private final AtomicReference ref; + + CasEntityStore(TableEntity initial) { + this.ref = new AtomicReference<>(new Versioned(0L, initial)); + } + + @Override + @SuppressWarnings("unchecked") + public E get( + NameIdentifier ident, EntityType entityType, Class type) { + return (E) ref.get().entity; + } + + @Override + @SuppressWarnings("unchecked") + public E update( + NameIdentifier ident, Class type, EntityType entityType, Function updater) + throws IOException { + Versioned base = ref.get(); + E updated = updater.apply((E) base.entity); + Versioned next = new Versioned(base.version + 1, (TableEntity) updated); + if (ref.compareAndSet(base, next)) { + return updated; + } + throw new IOException("Failed to update the entity: " + ident); + } + + // --- unused surface --------------------------------------------------- + + @Override + public void initialize(Config config) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean exists(NameIdentifier ident, EntityType entityType) { + throw new UnsupportedOperationException(); + } + + @Override + public void put(E e, boolean overwritten) { + throw new UnsupportedOperationException(); + } + + @Override + public List batchGet( + List idents, EntityType entityType, Class clazz) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean delete(NameIdentifier ident, EntityType entityType, boolean cascade) { + throw new UnsupportedOperationException(); + } + + @Override + public int batchDelete( + List> entitiesToDelete, boolean cascade) { + throw new UnsupportedOperationException(); + } + + @Override + public void batchPut(List entities, boolean overwritten) { + throw new UnsupportedOperationException(); + } + + @Override + public R executeInTransaction(Executable executable) { + throw new UnsupportedOperationException(); + } + + @Override + public void close() {} + } +} From 8ee3aa67fe4920a318aa737625da6faa93a484cb Mon Sep 17 00:00:00 2001 From: yuqi Date: Mon, 6 Jul 2026 19:26:53 +0800 Subject: [PATCH 5/5] [#11891] fix(lance): address CAS-retry review comments (jitter, shared constant, context, test) - Add a short randomized backoff (10-100ms) between lost-CAS retries so racing loads de-sync instead of immediately re-colliding (jerryshao). - Match the conflict via a shared TableMetaService.UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX constant instead of a duplicated literal string. - On exhausted retries, throw a contextual IOException (table + attempt count) chaining the last conflict, instead of rethrowing a bare conflict. - Keep the concurrency stress test cheap and non-flaky: fewer iterations, a @Timeout, and the retry backoff neutralized via a package-private seam. Signed-off-by: yuqi --- .../lakehouse/lance/LanceTableOperations.java | 47 +++++++++++++++++-- .../TestLanceConcurrentRepairStress.java | 13 +++-- .../relational/service/TableMetaService.java | 12 ++++- 3 files changed, 64 insertions(+), 8 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 4aab2d6b29e..7edfc468d53 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 @@ -31,6 +31,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.ThreadLocalRandom; import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -64,6 +65,7 @@ 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; @@ -86,6 +88,12 @@ public class LanceTableOperations extends ManagedTableOperations { */ private static final int REPAIR_UPDATE_MAX_ATTEMPTS = 5; + /** Lower bound (inclusive) of the randomized backoff slept between lost-CAS retries. */ + private static final long REPAIR_RETRY_MIN_BACKOFF_MS = 10; + + /** Upper bound (inclusive) of the randomized backoff slept between lost-CAS retries. */ + private static final long REPAIR_RETRY_MAX_BACKOFF_MS = 100; + public enum CreationMode { CREATE, EXIST_OK, @@ -556,8 +564,11 @@ private Table repairTableMetadata(NameIdentifier ident, Column[] columns, long d * 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}. Because the updater is idempotent, the loser re-reads the latest - * (already repaired) entity and retries instead of failing the whole load with a fatal error. + * 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. */ private TableEntity updateTableWithCasRetry( NameIdentifier ident, Function updater) throws IOException { @@ -569,7 +580,8 @@ private TableEntity updateTableWithCasRetry( // 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("Failed to update the entity:")) { + if (message == null + || !message.startsWith(TableMetaService.UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX)) { throw e; } @@ -581,9 +593,36 @@ private TableEntity updateTableWithCasRetry( REPAIR_UPDATE_MAX_ATTEMPTS, attempt < REPAIR_UPDATE_MAX_ATTEMPTS ? "retrying" : "retries exhausted", e); + + if (attempt < REPAIR_UPDATE_MAX_ATTEMPTS) { + backoffBeforeRetry(ident); + } } } - throw lastConflict; + throw new IOException( + String.format( + "Failed to update table %s after %d optimistic-lock retries", + ident, REPAIR_UPDATE_MAX_ATTEMPTS), + lastConflict); + } + + /** + * Sleeps a short randomized backoff between two lost-CAS retries so that racing loads de-sync + * instead of immediately colliding again. Package-private so tests can neutralize the sleep. + * + * @param ident the table being updated, used only for the interrupt error message + * @throws IOException if the thread is interrupted while backing off + */ + void backoffBeforeRetry(NameIdentifier ident) throws IOException { + long backoffMillis = + ThreadLocalRandom.current() + .nextLong(REPAIR_RETRY_MIN_BACKOFF_MS, REPAIR_RETRY_MAX_BACKOFF_MS + 1); + try { + Thread.sleep(backoffMillis); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while retrying metadata update for " + ident, ie); + } } private Table recordCheckedEmptyVersion(NameIdentifier ident, long datasetVersion) { 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 e2fa8a15931..4a67be530b5 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 @@ -58,6 +58,7 @@ import org.apache.gravitino.utils.PrincipalUtils; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import org.junit.jupiter.api.io.TempDir; import org.lance.Dataset; import org.mockito.Mockito; @@ -83,11 +84,14 @@ public class TestLanceConcurrentRepairStress { private static final NameIdentifier IDENT = NameIdentifier.of("schema", "table"); @Test + @Timeout(60) public void testConcurrentRepairLoadsSurviveCasContention() throws Exception { // The reported race is "two loads repair the same table at once". Also drive a small herd to - // confirm the bounded CAS retry stays robust beyond the minimal two-load case. - runStress(2, 3000); - runStress(8, 2000); + // confirm the bounded CAS retry stays robust beyond the minimal two-load case. Iterations are + // kept modest (and the retry backoff is neutralized in runStress) so this stays a fast, + // non-flaky unit test rather than a soak test — the pre-fix bug already fails on iteration 0. + runStress(2, 500); + runStress(8, 300); } private void runStress(int concurrency, int iterations) throws Exception { @@ -123,6 +127,9 @@ private void runStress(int concurrency, int iterations) throws Exception { "http://endpoint")); CasEntityStore store = new CasEntityStore(stale); LanceTableOperations ops = spy(new LanceTableOperations(store, schemaOps, idGenerator)); + // Neutralize the inter-retry backoff: this test validates the CAS re-read/retry recovery, + // not the sleep duration, and real sleeps would make it slow and timing-flaky. + Mockito.doNothing().when(ops).backoffBeforeRetry(Mockito.any()); Dataset dataset = mock(Dataset.class); when(dataset.version()).thenReturn(8L); when(dataset.getSchema()).thenReturn(datasetSchema); 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 72391bc31be..17ee0d5c3b2 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 @@ -56,6 +56,16 @@ /** 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 ops; @@ -236,7 +246,7 @@ public TableEntity updateTable( if (updateResult.get() > 0) { return newTableEntity; } else { - throw new IOException("Failed to update the entity: " + identifier); + throw new IOException(UPDATE_ENTITY_CONFLICT_MESSAGE_PREFIX + identifier); } }