jerryshao commented on code in PR #13487:
URL: https://github.com/apache/gravitino/pull/13487#discussion_r4236206102
##########
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:
[Important] This infers a collation collision from the absence of a live row
after the rollback, and that inference is wrong whenever the concurrent
creator's row is gone again by the time of this read.
Interleaving: we snapshot and see no `x`; another writer inserts `x` and
commits; our insert hits the unique key and the batch rolls back; that writer
(or a user) drops `x` and commits. The soft delete sets `deleted_at` to a
non-zero value, so `liveStatistics` now returns nothing, `anyMatch` is false,
and a purely transient race is reported as `IllegalStatisticNameException` -
HTTP 400 via ExceptionHandlers.java:1153, a permanent client error - with a
message about trailing spaces. A retry would have succeeded.
This is not MySQL-specific. On H2 and PostgreSQL there is no padding
collation at all, so a unique-key violation on this insert can *only* mean a
concurrent creator; reaching this branch there is wrong by construction.
The fallback is also removable rather than fixable in place, because both
real collation cases are already decidable without it: a request whose name
collides with a stored row is caught by the pre-check at lines 128-135, and an
in-request collision (`x` plus `x `) can be detected deterministically before
the insert by comparing the requested names with trailing spaces stripped. With
that check added, any remaining duplicate key is a concurrent creator and
belongs in `statisticConflict` as a 409. That also keeps the retry-storm case
the current code guards against: `{x, x }` both new still fails fast with 400
instead of looping on 409, since nothing is committed for a retry to observe.
Verified by: reading StatisticMetaService.java:136-168 and 291-318 in full
against the unique key in scripts/mysql/schema-2.0.0-mysql.sql:440 (which
includes `deleted_at`, so a soft-deleted row no longer collides) and the
soft-delete expression at StatisticBaseSQLProvider.java:229-231; the 400
mapping is ExceptionHandlers.java:1153-1154 with
IllegalStatisticNameException.java:25.
testBatchStatementConcurrencyFailuresAreConflicts reaches this branch only with
an injected duplicate and no concurrent writer, so the misclassification is
untested.
##########
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:
[Question] Does this cover an H2 lock-wait timeout?
The Javadoc just above says H2 reports 40001, which matches its deadlock
code, but a lock-wait timeout on H2 is a different error (`LOCK_TIMEOUT_1`,
SQLState HYT00) and would match neither the SQLStates nor the MySQL error codes
here - so it would surface as a 500 `PersistenceException` while the MySQL 1205
and PostgreSQL equivalents become 409. Since H2 is the backend these tests run
against most often and `testBatchStatementConcurrencyFailuresAreConflicts`
injects synthetic SQLExceptions rather than provoking a real timeout, the gap
would not show up in CI.
I could not verify H2's code in this run - the H2 jar is not resolvable in
this sandbox (libs.versions.toml:110 pins 2.2.224), and the repo's own H2
converter only enumerates 23505 and 22001 (H2ExceptionConverter.java:30-34). If
you have confirmed that a lock wait on H2 arrives as 40001, ignore this;
otherwise adding H2's lock-timeout code alongside the MySQL ones would close it.
Verified by: reading StatisticMetaService.java:364-385 and
H2ExceptionConverter.java:36-51, and confirming the test at
TestStatisticMetaService.java only constructs SQLExceptions by hand.
##########
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:
[Nit] This message names the stored statistic but not the requested name
that was rejected.
A batch can carry many names, and the caller sees only `'name'` plus the
target - it has to work out for itself which of the names it sent is the padded
alias. Including the offending requested name (the one in `names` that is
padding-equal to `stored`) would make the 400 actionable without guesswork. The
companion message at lines 313-317 does list the names.
Verified by: reading StatisticMetaService.java:119-135 - `stored` comes from
the keys of the backend read, while the rejected name is only ever in the local
`names` set and never reaches the message.
--
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]
