yuqi1129 commented on code in PR #13487:
URL: https://github.com/apache/gravitino/pull/13487#discussion_r4237069611
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java:
##########
@@ -100,11 +189,199 @@ public int batchDeleteStatisticPOs(
if (statisticNames == null || statisticNames.isEmpty()) {
return 0;
}
- Long entityId = EntityIdService.getEntityId(identifier, type);
+ NamespacedEntityId observed = EntityIdService.getEntityIds(identifier,
type);
+ Set<String> orderedNames =
+ statisticNames.stream()
+ .filter(Objects::nonNull)
+ .collect(Collectors.toCollection(TreeSet::new));
+ if (orderedNames.isEmpty()) {
+ return 0;
+ }
+ Map<String, StatisticPO> previous =
+ listStatisticPOs(observed, new ArrayList<>(orderedNames)).stream()
+ .collect(Collectors.toMap(StatisticPO::getStatisticName,
Function.identity()));
+ if (previous.isEmpty()) {
+ return 0;
+ }
+ // Look up by the exact requested names. A collation that ignores trailing
spaces can return the
+ // row of "name" for a request of "name "; dropping "name " must not drop
"name".
+ List<StatisticPO> observedRows =
+ orderedNames.stream()
+ .map(previous::get)
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ if (observedRows.isEmpty()) {
+ return 0;
+ }
+ int[] deleted = new int[] {0};
+ runFencedTransaction(
+ () -> {
+ LiveEndpointService.lockLiveEndpoint(identifier, type, observed);
+ deleted[0] =
+ executeStatement(mapper ->
mapper.batchDeleteStatisticPOsWithVersion(observedRows));
+ if (deleted[0] == observedRows.size()) {
+ return;
+ }
+ // Some observed rows were not deleted. A name that is still live
was changed or
+ // replaced meanwhile, which is a conflict. Otherwise a concurrent
drop already removed
+ // it; like a drop of a missing name, that is not a conflict and is
simply not counted.
+ List<StatisticPO> live = liveStatistics(observed,
names(observedRows));
+ if (!live.isEmpty()) {
+ throw statisticConflict(null, names(live), identifier);
+ }
+ },
+ observedRows,
+ identifier);
+ return deleted[0];
+ }
+
+ @VisibleForTesting
+ List<StatisticPO> listStatisticPOs(NamespacedEntityId endpoint, List<String>
names) {
+ return SessionUtils.getWithoutCommit(
+ StatisticMetaMapper.class,
+ mapper ->
+ mapper.listStatisticPOsByNames(endpoint.namespaceIds()[0],
endpoint.entityId(), names));
+ }
+
+ /** Runs one batch statement in the current transaction and returns its row
count. */
+ @VisibleForTesting
+ int executeStatement(Function<StatisticMetaMapper, Integer> statement) {
+ return SessionUtils.getWithoutCommit(StatisticMetaMapper.class, statement);
+ }
+
+ /**
+ * Runs a fenced statistic transaction. A failure caused by a concurrent
writer anywhere in the
+ * transaction, including the fence's locking reads, is reported as a
conflict on {@code pos}.
+ */
+ private static void runFencedTransaction(
+ Runnable operations, List<StatisticPO> pos, NameIdentifier target) {
+ try {
+ SessionUtils.doMultipleWithCommit(operations);
+ } catch (RuntimeException e) {
+ throw translateConflict(e, pos, target);
+ }
+ }
- return SessionUtils.doWithCommitAndFetchResult(
+ /**
+ * Converts a failure caused by a concurrent writer into a conflict. A
duplicate insert means a
+ * statistic was created after the snapshot. A deadlock, serialization
failure or lock wait
+ * timeout can happen because a batch statement locks its rows in the order
of the database's
+ * plan, which differs between batches. The whole batch is rolled back
either way, so these are
+ * reported as a conflict rather than an internal server error.
+ */
+ private static RuntimeException translateConflict(
+ RuntimeException failure, List<StatisticPO> pos, NameIdentifier target) {
+ if (failure instanceof OptimisticLockException) {
+ return failure;
+ }
+ if (isDuplicateKey(failure)) {
+ return statisticConflict(failure, names(pos), target);
+ }
+ if (isConcurrencyFailure(failure)) {
+ return new OptimisticLockException(
+ failure,
+ "The statistics %s of %s could not be changed because of a
concurrent operation"
+ + " (deadlock or lock wait timeout); retry the operation",
+ names(pos),
+ target);
+ }
+ return failure;
+ }
+
+ /**
+ * Distinguishes a duplicate insert caused by the backend collation from a
concurrent creator. A
+ * collation that ignores trailing spaces (MySQL) rejects "name " in the
same request as "name",
+ * or next to an existing "name", on every attempt. If no live row has
exactly one of the inserted
+ * names after the rollback, nobody else created them, so the names are
illegal for this backend.
+ */
+ private static RuntimeException collationCollisionOr(
+ OptimisticLockException conflict,
+ List<StatisticPO> inserts,
+ NamespacedEntityId endpoint,
+ NameIdentifier target) {
+ if (inserts.isEmpty() || !isDuplicateKey(conflict)) {
+ return conflict;
+ }
+ Set<String> live =
+ liveStatistics(endpoint, names(inserts)).stream()
+ .map(StatisticPO::getStatisticName)
+ .collect(Collectors.toSet());
+ if (inserts.stream().anyMatch(po -> live.contains(po.getStatisticName())))
{
+ return conflict;
+ }
Review Comment:
Fixed in 212b8f151c: the post-rollback inference is removed.
- On a backend whose statistic key ignores trailing spaces (only MySQL;
`StatisticSQLProviderFactory.namesIgnoreTrailingSpaces()`), names in one
request that are equal after stripping trailing spaces are rejected up front
with `IllegalStatisticNameException` (400), before any statement runs.
- Collisions with a stored row are still caught by the existing pre-check.
- Any duplicate key that reaches the insert is now a concurrent creator and
becomes a 409 through `translateConflict`, also when that row is gone again
later.
- `{x, x }` still fails fast with 400 on MySQL. H2 and PostgreSQL keep
accepting both names, as before.
- `testBatchStatementConcurrencyFailuresAreConflicts` now asserts that a
duplicate key on a write is an `OptimisticLockException`; it previously
asserted `IllegalStatisticNameException`.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java:
##########
@@ -100,11 +189,199 @@ public int batchDeleteStatisticPOs(
if (statisticNames == null || statisticNames.isEmpty()) {
return 0;
}
- Long entityId = EntityIdService.getEntityId(identifier, type);
+ NamespacedEntityId observed = EntityIdService.getEntityIds(identifier,
type);
+ Set<String> orderedNames =
+ statisticNames.stream()
+ .filter(Objects::nonNull)
+ .collect(Collectors.toCollection(TreeSet::new));
+ if (orderedNames.isEmpty()) {
+ return 0;
+ }
+ Map<String, StatisticPO> previous =
+ listStatisticPOs(observed, new ArrayList<>(orderedNames)).stream()
+ .collect(Collectors.toMap(StatisticPO::getStatisticName,
Function.identity()));
+ if (previous.isEmpty()) {
+ return 0;
+ }
+ // Look up by the exact requested names. A collation that ignores trailing
spaces can return the
+ // row of "name" for a request of "name "; dropping "name " must not drop
"name".
+ List<StatisticPO> observedRows =
+ orderedNames.stream()
+ .map(previous::get)
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ if (observedRows.isEmpty()) {
+ return 0;
+ }
+ int[] deleted = new int[] {0};
+ runFencedTransaction(
+ () -> {
+ LiveEndpointService.lockLiveEndpoint(identifier, type, observed);
+ deleted[0] =
+ executeStatement(mapper ->
mapper.batchDeleteStatisticPOsWithVersion(observedRows));
+ if (deleted[0] == observedRows.size()) {
+ return;
+ }
+ // Some observed rows were not deleted. A name that is still live
was changed or
+ // replaced meanwhile, which is a conflict. Otherwise a concurrent
drop already removed
+ // it; like a drop of a missing name, that is not a conflict and is
simply not counted.
+ List<StatisticPO> live = liveStatistics(observed,
names(observedRows));
+ if (!live.isEmpty()) {
+ throw statisticConflict(null, names(live), identifier);
+ }
+ },
+ observedRows,
+ identifier);
+ return deleted[0];
+ }
+
+ @VisibleForTesting
+ List<StatisticPO> listStatisticPOs(NamespacedEntityId endpoint, List<String>
names) {
+ return SessionUtils.getWithoutCommit(
+ StatisticMetaMapper.class,
+ mapper ->
+ mapper.listStatisticPOsByNames(endpoint.namespaceIds()[0],
endpoint.entityId(), names));
+ }
+
+ /** Runs one batch statement in the current transaction and returns its row
count. */
+ @VisibleForTesting
+ int executeStatement(Function<StatisticMetaMapper, Integer> statement) {
+ return SessionUtils.getWithoutCommit(StatisticMetaMapper.class, statement);
+ }
+
+ /**
+ * Runs a fenced statistic transaction. A failure caused by a concurrent
writer anywhere in the
+ * transaction, including the fence's locking reads, is reported as a
conflict on {@code pos}.
+ */
+ private static void runFencedTransaction(
+ Runnable operations, List<StatisticPO> pos, NameIdentifier target) {
+ try {
+ SessionUtils.doMultipleWithCommit(operations);
+ } catch (RuntimeException e) {
+ throw translateConflict(e, pos, target);
+ }
+ }
- return SessionUtils.doWithCommitAndFetchResult(
+ /**
+ * Converts a failure caused by a concurrent writer into a conflict. A
duplicate insert means a
+ * statistic was created after the snapshot. A deadlock, serialization
failure or lock wait
+ * timeout can happen because a batch statement locks its rows in the order
of the database's
+ * plan, which differs between batches. The whole batch is rolled back
either way, so these are
+ * reported as a conflict rather than an internal server error.
+ */
+ private static RuntimeException translateConflict(
+ RuntimeException failure, List<StatisticPO> pos, NameIdentifier target) {
+ if (failure instanceof OptimisticLockException) {
+ return failure;
+ }
+ if (isDuplicateKey(failure)) {
+ return statisticConflict(failure, names(pos), target);
+ }
+ if (isConcurrencyFailure(failure)) {
+ return new OptimisticLockException(
+ failure,
+ "The statistics %s of %s could not be changed because of a
concurrent operation"
+ + " (deadlock or lock wait timeout); retry the operation",
+ names(pos),
+ target);
+ }
+ return failure;
+ }
+
+ /**
+ * Distinguishes a duplicate insert caused by the backend collation from a
concurrent creator. A
+ * collation that ignores trailing spaces (MySQL) rejects "name " in the
same request as "name",
+ * or next to an existing "name", on every attempt. If no live row has
exactly one of the inserted
+ * names after the rollback, nobody else created them, so the names are
illegal for this backend.
+ */
+ private static RuntimeException collationCollisionOr(
+ OptimisticLockException conflict,
+ List<StatisticPO> inserts,
+ NamespacedEntityId endpoint,
+ NameIdentifier target) {
+ if (inserts.isEmpty() || !isDuplicateKey(conflict)) {
+ return conflict;
+ }
+ Set<String> live =
+ liveStatistics(endpoint, names(inserts)).stream()
+ .map(StatisticPO::getStatisticName)
+ .collect(Collectors.toSet());
+ if (inserts.stream().anyMatch(po -> live.contains(po.getStatisticName())))
{
+ return conflict;
+ }
+ return new IllegalStatisticNameException(
+ conflict.getCause(),
+ "Statistic names %s of %s collide with each other or with an existing
statistic in the"
+ + " backend collation; names must not differ only in trailing
spaces",
+ names(inserts),
+ target);
+ }
+
+ private static OptimisticLockException statisticConflict(
+ @Nullable Throwable cause, List<String> names, NameIdentifier target) {
+ return new OptimisticLockException(
+ cause,
+ "One or more of the statistics %s of %s were changed concurrently;
retry the operation",
+ names,
+ target);
+ }
+
+ private static List<StatisticPO> liveStatistics(NamespacedEntityId endpoint,
List<String> names) {
+ return SessionUtils.getWithoutCommit(
StatisticMetaMapper.class,
- mapper -> mapper.batchDeleteStatisticPOs(entityId, statisticNames));
+ mapper ->
+ mapper.listStatisticPOsByNames(endpoint.namespaceIds()[0],
endpoint.entityId(), names));
+ }
+
+ private static List<String> names(List<StatisticPO> pos) {
+ return
pos.stream().map(StatisticPO::getStatisticName).sorted().collect(Collectors.toList());
+ }
+
+ /** Returns a PO that identifies the observed row and carries the
replacement value. */
+ private static StatisticPO replacementOf(StatisticPO observed, StatisticPO
replacement) {
+ return StatisticPO.builder()
+ .withStatisticId(observed.getStatisticId())
+ .withStatisticName(observed.getStatisticName())
+ .withMetalakeId(observed.getMetalakeId())
+ .withMetadataObjectId(observed.getMetadataObjectId())
+ .withMetadataObjectType(observed.getMetadataObjectType())
+ .withCurrentVersion(observed.getCurrentVersion())
+ .withLastVersion(observed.getLastVersion())
+ .withDeletedAt(observed.getDeletedAt())
+ .withStatisticValue(replacement.getStatisticValue())
+ .withAuditInfo(replacement.getAuditInfo())
+ .build();
+ }
+
+ /**
+ * Returns whether a failure is a unique-key violation. PostgreSQL and H2
report SQLState 23505,
+ * and MySQL reports error code 1062, the same codes the backend exception
converters match.
+ */
+ private static boolean isDuplicateKey(Throwable failure) {
+ return hasSqlException(failure, Set.of("23505"), Set.of(1062));
+ }
+
+ /**
+ * Returns whether a failure is a deadlock, serialization failure or lock
wait timeout caused by a
+ * concurrent writer. PostgreSQL reports SQLState 40001 or 40P01 and H2
reports 40001. MySQL
+ * reports error code 1213 for a deadlock and 1205 for a lock wait timeout;
both are matched by
+ * error code because the SQLState of a lock wait timeout depends on the
driver.
+ */
+ private static boolean isConcurrencyFailure(Throwable failure) {
+ return hasSqlException(failure, Set.of("40001", "40P01"), Set.of(1213,
1205));
Review Comment:
Good catch, it did not. A lock-wait timeout on H2 is error code 50200 with
SQLState HYT00, not 40001. Fixed in 212b8f151c:
- `isConcurrencyFailure` now matches H2 50200, and also PostgreSQL 55P03
(`lock_not_available`).
- The vendor codes are named constants.
- The synthetic list in `testBatchStatementConcurrencyFailuresAreConflicts`
covers both.
- New `testH2LockWaitTimeoutIsConflict` holds the target table row `FOR
UPDATE` from another session, so the fenced write hits a real H2 lock timeout.
The test asserts a 409 whose cause chain carries error code 50200.
- Removing 50200 from the set fails both tests.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java:
##########
@@ -79,19 +108,79 @@ public void batchInsertStatisticPOsOnDuplicateKeyUpdate(
namespacedEntityId.namespaceIds()[0],
namespacedEntityId.entityId(),
NameIdentifierUtil.toMetadataObject(entity, type).type());
- // Statistics have their own write API, so they do not inherit the schema
fence from the
- // metadata object update path. Fence schema-scoped targets explicitly to
keep a schema cascade
- // from deleting the target and then missing this independently committed
statistic upsert.
- doWithSchemaWriteLockIfNeeded(
- entity,
- type,
- namespacedEntityId,
- () ->
- SessionUtils.doWithoutCommit(
- StatisticMetaMapper.class,
- mapper ->
mapper.batchInsertStatisticPOsOnDuplicateKeyUpdate(pos)));
+ Set<String> names = new HashSet<>();
+ for (StatisticPO po : pos) {
+ if (!names.add(po.getStatisticName())) {
+ throw new IllegalArgumentException(
+ "Duplicate statistic name in batch: " + po.getStatisticName());
+ }
+ }
+ pos.sort(Comparator.comparing(StatisticPO::getStatisticName));
+ Map<String, StatisticPO> previous =
+ listStatisticPOs(
+ namespacedEntityId,
+
pos.stream().map(StatisticPO::getStatisticName).collect(Collectors.toList()))
+ .stream()
+ .collect(Collectors.toMap(StatisticPO::getStatisticName,
Function.identity()));
+ // A collation that ignores trailing spaces (MySQL) returns the row of
"name" for "name ", and
+ // its unique key treats them as the same statistic. Writing "name " can
never succeed there, so
+ // it is rejected as an illegal name instead of a conflict that suggests a
retry.
+ for (String stored : previous.keySet()) {
+ if (!names.contains(stored)) {
+ throw new IllegalStatisticNameException(
+ "Statistic name is equivalent to the existing statistic '%s' of %s
in the backend"
+ + " collation; names must not differ only in trailing spaces",
+ stored, entity);
Review Comment:
Fixed in 212b8f151c: the message now names the requested name and the stored
statistic it collides with: `Statistic name 'name ' is equivalent to the
existing statistic 'name' ...`. The new in-request rejection names both
requested names. The two trailing-space tests assert both names on MySQL.
--
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]
