jerryshao commented on code in PR #13737:
URL: https://github.com/apache/gravitino/pull/13737#discussion_r4245411498
##########
core/src/main/java/org/apache/gravitino/stats/StatisticManager.java:
##########
@@ -200,14 +196,7 @@ public void updateStatistics(
.build();
statisticEntities.add(statistic);
}
- TreeLockUtils.doWithTreeLock(
- identifier,
- LockType.WRITE,
- (Executable<Void, IOException>)
- () -> {
- store.batchPut(statisticEntities, true);
- return null;
- });
+ store.batchPut(statisticEntities, true);
Review Comment:
[Question] Removing this lock makes a single server's table drop contend for
the table row, and on that side a lock-wait timeout is not translated into a
conflict.
Before this commit the two could not overlap on one server: this write took
`WRITE` on the table node, which takes `READ` on the schema node
(`TreeLock.java:100-106`), while `TableOperationDispatcher#dropTable` takes
`WRITE` on the schema node (`TableOperationDispatcher.java:413-417`). Now they
meet in the database instead. The statistic transaction holds the table row
through `lockLiveEndpoint`, which is `FOR UPDATE` on H2
(`LiveEndpointSQLProvider.java:112-116`), and the drop's first statement
updates that same row (`TableMetaService.java:333-341`), so the drop queues
behind every in-flight statistic write.
The asymmetry is in what happens when that wait expires. The statistic side
turns a deadlock or lock-wait timeout into `OptimisticLockException`, i.e. 409
(`StatisticMetaService.java:286-305`); nothing does that for the table-delete
path - `isConcurrencyFailure` exists only in `StatisticMetaService` - so a drop
that loses the wait surfaces as a raw store failure, i.e. 500. The shipped
default backend is plain `jdbc:h2` (`Configs.java:93`), and the repo's own note
is that H2 gives new sessions a one-second lock timeout
(`TestOwnerAssignmentWrites.java:455-457`); the new race test raises it to 30 s
precisely so the drop never loses the wait, which is also the evidence that it
can.
Not necessarily work for this PR, but worth a decision: either translate
concurrency failures on the table-delete path too, or add one line to the
user-facing section saying that on a single server a table drop racing
table-statistic writes can now fail on a lock-wait timeout, as it already could
across servers since #13487.
Verified by: reading the removed lock here and `TreeLock#lock`, then
`dropTable`, then the `FOR UPDATE` clause the statistic fence emits and the
drop's first statement; `grep -rn isConcurrencyFailure core/src/main` returns
only `StatisticMetaService`.
##########
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:
[Nit] `SET DEFAULT_LOCK_TIMEOUT` only reaches sessions opened after it, so
one pooled connection in this race can still carry H2's one-second timeout.
`TestJDBCBackend#init` opens a session from the shared `SqlSessionFactory`
to truncate the tables (`TestJDBCBackend.java:113-116`) before this method
runs, and closing it returns the connection, i.e. the H2 session, to the pool
with the old timeout. A race thread that picks it up keeps one second.
`TestOwnerAssignmentWrites` handles exactly this by also issuing `SET
LOCK_TIMEOUT 30000` on each session it controls
(`TestOwnerAssignmentWrites.java:426-440`) besides raising the database default.
It matters most in `testUpsertsRacingTableDropLeaveNoStatistics`: a drop
that times out is neither `Boolean.TRUE` nor an accepted outcome, so that round
fails rather than flagging a conflict. Worth either setting the session timeout
inside `upsert`/the drop action, or putting `LOCK_TIMEOUT=30000` in this
class's H2 URL as `TestTagManager.java:180` and `TestPolicyManager.java:253` do.
Verified by: reading `raiseDefaultLockTimeout` and
`TestJDBCBackend#init`/`truncateAllTables` on this head, then the two existing
patterns in `TestOwnerAssignmentWrites` and the manager race tests.
##########
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:
[Nit] This invariant cannot fail, so the round's end state is effectively
unchecked.
`uk_si_mi_mo_del (statistic_name, metadata_object_id, deleted_at)` already
makes two live rows with one name impossible
(`scripts/h2/schema-1.2.0-h2.sql:397`), so `distinct().count() == size()` holds
whatever the race did.
Every operation here is all-or-nothing - an upsert writes all five names in
one transaction, and a drop that deleted only part of what it observed throws
and rolls back (`StatisticMetaService.java:233-245`) - so the round should end
with either exactly `NAMES` or nothing live. Asserting that, rather than
distinctness, would catch a partially applied batch, which is the failure this
test exists to detect.
Verified by: reading the H2, MySQL and PostgreSQL schema files for the
unique key, then `batchDeleteStatisticPOs` and `writeStatisticsWithVersion` to
confirm each batch is one transaction.
--
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]
