yuqi1129 commented on code in PR #13737: URL: https://github.com/apache/gravitino/pull/13737#discussion_r4245767621
########## core/src/test/java/org/apache/gravitino/storage/relational/service/TestStatisticConcurrentWrites.java: ########## @@ -0,0 +1,235 @@ +/* + * 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.service; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.time.Instant; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.gravitino.Entity; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; +import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.StatisticEntity; +import org.apache.gravitino.meta.TableEntity; +import org.apache.gravitino.meta.TableStatisticEntity; +import org.apache.gravitino.stats.StatisticValues; +import org.apache.gravitino.storage.RandomIdGenerator; +import org.apache.gravitino.storage.relational.TestJDBCBackend; +import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; +import org.apache.gravitino.utils.RaceTestUtils; +import org.apache.ibatis.session.SqlSession; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.TestTemplate; + +/** + * Concurrent table-level statistic writes, which do not take the table tree lock. Every outcome + * must be a success or a conflict the caller can retry; a deadlock must never surface as an + * internal error, and no statistic may outlive its table. + */ +public class TestStatisticConcurrentWrites extends TestJDBCBackend { + + private static final String METALAKE = "metalake_statistic_race"; + private static final String CATALOG = "catalog_statistic_race"; + private static final String SCHEMA = "schema_statistic_race"; + private static final List<String> NAMES = ImmutableList.of("s0", "s1", "s2", "s3", "s4"); + private static final int ROUNDS = 20; + + private final AuditInfo auditInfo = + AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build(); + + @TestTemplate + public void testOverlappingUpsertsInOppositeOrderSucceedOrConflict() throws Exception { + raiseDefaultLockTimeout(); + createParentEntities(METALAKE, CATALOG, SCHEMA, auditInfo); + TableEntity table = createTable("table_upsert_race"); + + for (int round = 0; round < ROUNDS; round++) { + AtomicInteger writer = new AtomicInteger(); + long base = round * 100L; + List<Object> outcomes = + RaceTestUtils.runTogether( + 8, + () -> { + int index = writer.getAndIncrement(); + upsert(table, index % 2 == 0 ? NAMES : Lists.reverse(NAMES), base + index); + return null; + }); + + // The version check lets one writer of a statistic win and reports the others as conflicts. + // Locking the rows in name order keeps opposite request orders from deadlocking. + Assertions.assertTrue(outcomes.contains(null), "One writer must win: " + outcomes); + for (Object outcome : outcomes) { + Assertions.assertTrue( + outcome == null || outcome instanceof OptimisticLockException, + "Unexpected outcome: " + outcome); + } + Assertions.assertEquals(NAMES, liveNames(table)); + } + } + + @TestTemplate + public void testUpsertsRacingStatisticDropsSucceedOrConflict() throws Exception { + raiseDefaultLockTimeout(); + createParentEntities(METALAKE, CATALOG, SCHEMA, auditInfo); + TableEntity table = createTable("table_drop_race"); + + for (int round = 0; round < ROUNDS; round++) { + upsert(table, NAMES, round); + AtomicInteger worker = new AtomicInteger(); + long value = round * 100L; + List<Object> outcomes = + RaceTestUtils.runTogether( + 8, + () -> { + int index = worker.getAndIncrement(); + if (index % 2 == 0) { + upsert(table, Lists.reverse(NAMES), value + index); + return null; + } + return StatisticMetaService.getInstance() + .batchDeleteStatisticPOs( + table.nameIdentifier(), Entity.EntityType.TABLE, NAMES); + }); + + for (Object outcome : outcomes) { + Assertions.assertTrue( + outcome == null + || outcome instanceof Integer + || outcome instanceof OptimisticLockException, + "Unexpected outcome: " + outcome); + } + // Whatever won, each name has at most one live row. + List<String> live = liveNames(table); + Assertions.assertEquals(live.stream().distinct().count(), live.size()); + } + } + + @TestTemplate + public void testUpsertsRacingTableDropLeaveNoStatistics() throws Exception { + raiseDefaultLockTimeout(); + createParentEntities(METALAKE, CATALOG, SCHEMA, auditInfo); + + for (int round = 0; round < ROUNDS; round++) { + TableEntity table = createTable("table_cascade_race_" + round); + // Insert in reverse name order, so the row ids of the cascade delete and the names of the + // upserts order the statistic rows differently. + for (String name : Lists.reverse(NAMES)) { + upsert(table, ImmutableList.of(name), 0); + } + AtomicInteger worker = new AtomicInteger(); + List<Object> outcomes = + RaceTestUtils.runTogether( + 8, + (Callable<Object>) + () -> { + if (worker.getAndIncrement() == 0) { + return TableMetaService.getInstance().deleteTable(table.nameIdentifier()); + } + upsert(table, NAMES, 1); + return null; + }); + + Assertions.assertTrue( + outcomes.contains(Boolean.TRUE), "The table drop must win: " + outcomes); + for (Object outcome : outcomes) { + Assertions.assertTrue( + outcome == null + || Boolean.TRUE.equals(outcome) + || outcome instanceof NoSuchEntityException + || outcome instanceof OptimisticLockException, + "Unexpected outcome: " + outcome); + } + Assertions.assertEquals(0, countLiveStatistics(table.id())); + } + } + + private TableEntity createTable(String name) throws Exception { + TableEntity table = + createTableEntity( + RandomIdGenerator.INSTANCE.nextId(), + Namespace.of(METALAKE, CATALOG, SCHEMA), + name, + auditInfo); + backend.insert(table, false); + return table; + } + + private void upsert(TableEntity table, List<String> names, long value) { + List<StatisticEntity> statistics = + names.stream() + .map( + name -> + (StatisticEntity) + TableStatisticEntity.builder() + .withId(RandomIdGenerator.INSTANCE.nextId()) + .withName(name) + .withValue(StatisticValues.longValue(value)) + .withAuditInfo(auditInfo) + .build()) + .collect(Collectors.toList()); + StatisticMetaService.getInstance() + .writeStatisticsWithVersion(statistics, table.nameIdentifier(), Entity.EntityType.TABLE); + } + + private List<String> liveNames(TableEntity table) { + return StatisticMetaService.getInstance() + .listStatisticsByEntity(table.nameIdentifier(), Entity.EntityType.TABLE) + .stream() + .map(StatisticEntity::name) + .sorted() + .collect(Collectors.toList()); + } + + private int countLiveStatistics(long tableId) throws SQLException { + try (SqlSession session = + SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true); + Statement statement = session.getConnection().createStatement(); + ResultSet rs = + statement.executeQuery( + "SELECT COUNT(*) FROM statistic_meta WHERE metadata_object_id = " + + tableId + + " AND deleted_at = 0")) { + rs.next(); + return rs.getInt(1); + } + } + + /** + * H2 gives new sessions a short lock timeout, so writers that queue behind each other would fail + * on a slow machine. The other backends wait long enough by default. + */ + private void raiseDefaultLockTimeout() throws SQLException { + if (!"h2".equals(backendType)) { + return; + } + try (SqlSession session = + SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true); + Statement statement = session.getConnection().createStatement()) { + statement.execute("SET DEFAULT_LOCK_TIMEOUT 30000"); Review Comment: Fixed in bdf711da72. - `raiseDefaultLockTimeout` still raises the database default. - It now also borrows every connection that is idle in the pool at once, so each idle connection is handed out exactly once, and runs `SET LOCK_TIMEOUT 30000` on each before returning it. - The session that `TestJDBCBackend#init` returned to the pool therefore no longer carries the 1 s timeout. - `TestStatisticConcurrentWrites` passes 9/9 on H2, MySQL and PostgreSQL. ########## core/src/test/java/org/apache/gravitino/storage/relational/service/TestStatisticConcurrentWrites.java: ########## @@ -0,0 +1,235 @@ +/* + * 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.service; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.time.Instant; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.gravitino.Entity; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; +import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.StatisticEntity; +import org.apache.gravitino.meta.TableEntity; +import org.apache.gravitino.meta.TableStatisticEntity; +import org.apache.gravitino.stats.StatisticValues; +import org.apache.gravitino.storage.RandomIdGenerator; +import org.apache.gravitino.storage.relational.TestJDBCBackend; +import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; +import org.apache.gravitino.utils.RaceTestUtils; +import org.apache.ibatis.session.SqlSession; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.TestTemplate; + +/** + * Concurrent table-level statistic writes, which do not take the table tree lock. Every outcome + * must be a success or a conflict the caller can retry; a deadlock must never surface as an + * internal error, and no statistic may outlive its table. + */ +public class TestStatisticConcurrentWrites extends TestJDBCBackend { + + private static final String METALAKE = "metalake_statistic_race"; + private static final String CATALOG = "catalog_statistic_race"; + private static final String SCHEMA = "schema_statistic_race"; + private static final List<String> NAMES = ImmutableList.of("s0", "s1", "s2", "s3", "s4"); + private static final int ROUNDS = 20; + + private final AuditInfo auditInfo = + AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build(); + + @TestTemplate + public void testOverlappingUpsertsInOppositeOrderSucceedOrConflict() throws Exception { + raiseDefaultLockTimeout(); + createParentEntities(METALAKE, CATALOG, SCHEMA, auditInfo); + TableEntity table = createTable("table_upsert_race"); + + for (int round = 0; round < ROUNDS; round++) { + AtomicInteger writer = new AtomicInteger(); + long base = round * 100L; + List<Object> outcomes = + RaceTestUtils.runTogether( + 8, + () -> { + int index = writer.getAndIncrement(); + upsert(table, index % 2 == 0 ? NAMES : Lists.reverse(NAMES), base + index); + return null; + }); + + // The version check lets one writer of a statistic win and reports the others as conflicts. + // Locking the rows in name order keeps opposite request orders from deadlocking. + Assertions.assertTrue(outcomes.contains(null), "One writer must win: " + outcomes); + for (Object outcome : outcomes) { + Assertions.assertTrue( + outcome == null || outcome instanceof OptimisticLockException, + "Unexpected outcome: " + outcome); + } + Assertions.assertEquals(NAMES, liveNames(table)); + } + } + + @TestTemplate + public void testUpsertsRacingStatisticDropsSucceedOrConflict() throws Exception { + raiseDefaultLockTimeout(); + createParentEntities(METALAKE, CATALOG, SCHEMA, auditInfo); + TableEntity table = createTable("table_drop_race"); + + for (int round = 0; round < ROUNDS; round++) { + upsert(table, NAMES, round); + AtomicInteger worker = new AtomicInteger(); + long value = round * 100L; + List<Object> outcomes = + RaceTestUtils.runTogether( + 8, + () -> { + int index = worker.getAndIncrement(); + if (index % 2 == 0) { + upsert(table, Lists.reverse(NAMES), value + index); + return null; + } + return StatisticMetaService.getInstance() + .batchDeleteStatisticPOs( + table.nameIdentifier(), Entity.EntityType.TABLE, NAMES); + }); + + for (Object outcome : outcomes) { + Assertions.assertTrue( + outcome == null + || outcome instanceof Integer + || outcome instanceof OptimisticLockException, + "Unexpected outcome: " + outcome); + } + // Whatever won, each name has at most one live row. + List<String> live = liveNames(table); + Assertions.assertEquals(live.stream().distinct().count(), live.size()); Review Comment: Fixed in bdf711da72. - The drop race now asserts that each round ends with exactly `NAMES` or with nothing live, which catches a partially applied batch. - Separately, the table-drop race now asserts that no outcome carries a deadlock in its cause chain (SQLState 40001/40P01, MySQL 1213). This pins that the drop and the writers queue on the table row rather than deadlock and get mapped to 409. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
