jerryshao commented on code in PR #13487:
URL: https://github.com/apache/gravitino/pull/13487#discussion_r4244274024
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java:
##########
@@ -100,11 +200,202 @@ 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];
+ }
- return SessionUtils.doWithCommitAndFetchResult(
+ @VisibleForTesting
+ List<StatisticPO> listStatisticPOs(NamespacedEntityId endpoint, List<String>
names) {
+ return SessionUtils.getWithoutCommit(
StatisticMetaMapper.class,
- mapper -> mapper.batchDeleteStatisticPOs(entityId, statisticNames));
+ 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);
+ }
+ }
+
+ /**
+ * 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;
+ }
+
+ /**
+ * Rejects requested names that the backend key treats as one statistic
because they differ only
+ * in trailing spaces. Such a request can never succeed, so it fails as an
illegal name before any
+ * write instead of as a conflict that suggests a retry.
+ */
+ private static void rejectTrailingSpaceAliases(List<StatisticPO> pos,
NameIdentifier target) {
+ Map<String, String> requestedByKey = new HashMap<>();
+ for (StatisticPO po : pos) {
+ String name = po.getStatisticName();
+ String other = requestedByKey.putIfAbsent(withoutTrailingSpaces(name),
name);
+ if (other != null) {
+ throw new IllegalStatisticNameException(
+ "Statistic names '%s' and '%s' of %s are the same statistic in the
backend collation;"
+ + " names must not differ only in trailing spaces",
+ other, name, target);
+ }
+ }
+ }
+
+ private static String withoutTrailingSpaces(String name) {
+ int end = name.length();
+ while (end > 0 && name.charAt(end - 1) == ' ') {
+ end--;
+ }
+ return name.substring(0, end);
+ }
Review Comment:
[Nit] This is `StringUtils.stripEnd(name, " ")` from commons-lang3, which is
already on core's compile path (for example
core/src/main/java/org/apache/gravitino/catalog/TopicOperationDispatcher.java:28).
Same semantics - it strips every trailing U+0020 and returns the input
unchanged when there is none - in one line instead of six, and it names the
intent at both call sites. Not worth a push on its own; fold it in if you touch
this file again, or keep the local helper if you would rather not add the
import here.
Verified by: reading the method against `StringUtils.stripEnd` semantics,
and `grep -rn commons.lang3.StringUtils core/src/main/java` showing the
dependency is available to this module.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java:
##########
@@ -100,11 +200,202 @@ 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];
+ }
- return SessionUtils.doWithCommitAndFetchResult(
+ @VisibleForTesting
+ List<StatisticPO> listStatisticPOs(NamespacedEntityId endpoint, List<String>
names) {
+ return SessionUtils.getWithoutCommit(
StatisticMetaMapper.class,
- mapper -> mapper.batchDeleteStatisticPOs(entityId, statisticNames));
+ 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);
+ }
+ }
+
+ /**
+ * 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;
+ }
+
+ /**
+ * Rejects requested names that the backend key treats as one statistic
because they differ only
+ * in trailing spaces. Such a request can never succeed, so it fails as an
illegal name before any
+ * write instead of as a conflict that suggests a retry.
+ */
+ private static void rejectTrailingSpaceAliases(List<StatisticPO> pos,
NameIdentifier target) {
+ Map<String, String> requestedByKey = new HashMap<>();
+ for (StatisticPO po : pos) {
+ String name = po.getStatisticName();
+ String other = requestedByKey.putIfAbsent(withoutTrailingSpaces(name),
name);
+ if (other != null) {
+ throw new IllegalStatisticNameException(
+ "Statistic names '%s' and '%s' of %s are the same statistic in the
backend collation;"
+ + " names must not differ only in trailing spaces",
+ other, name, target);
+ }
+ }
+ }
+
+ private static String withoutTrailingSpaces(String name) {
+ int end = name.length();
+ while (end > 0 && name.charAt(end - 1) == ' ') {
+ end--;
+ }
+ return name.substring(0, end);
+ }
+
+ 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.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, or 55P03
for a lock timeout. H2
+ * reports 40001 for a deadlock and error code 50200 (SQLState HYT00) for a
lock wait timeout.
+ * MySQL reports error code 1213 for a deadlock and 1205 for a lock wait
timeout. Vendor error
+ * codes are matched because the SQLState of a lock wait timeout depends on
the driver.
Review Comment:
[Nit] Two of the three PostgreSQL codes this sentence attributes to
PostgreSQL cannot occur in Gravitino's own configuration, so the Javadoc reads
as fact where it is really a deliberately wide tolerance set.
Sessions are opened at READ COMMITTED (SqlSessions.java:65), and PostgreSQL
raises 40001 (serialization_failure) only under REPEATABLE READ or
SERIALIZABLE; at READ COMMITTED the code it does raise here is 40P01. 55P03
(lock_not_available) requires `NOWAIT` or a non-zero `lock_timeout`, and the
fence uses a plain `FOR SHARE` (LiveEndpointSQLProvider.java:122) with no
`lock_timeout` set anywhere in the repository - though a DBA can set it
server-side, which is a good reason to keep tolerating the code.
Suggest saying the set is intentionally wider than what each backend emits
today (40001 is H2's deadlock code; 55P03 covers a DBA-configured
`lock_timeout`), so a later cleanup does not trim 40001 or 55P03 on the grounds
that PostgreSQL does not report them here.
Verified by: SqlSessions.java:65 for the isolation level,
LiveEndpointSQLProvider.java:115-122 for the three lock clauses (no NOWAIT),
and `grep -rni "lock_timeout|NOWAIT" core/src/main/java scripts/postgresql`
matching only the new `H2_LOCK_TIMEOUT` constant.
--
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]
