yuqi1129 commented on code in PR #13487:
URL: https://github.com/apache/gravitino/pull/13487#discussion_r4244330766
##########
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:
Fixed in
https://github.com/apache/gravitino/commit/036f5077f2736114adbba548a1f29705524a0f38.
Replaced the helper with `StringUtils.stripEnd(name, " ")` at all three
normalization sites and removed the helper. This preserves stripping only
trailing U+0020 spaces. The real H2/MySQL/PostgreSQL service suite passed (58
passed, 2 H2-only timeout skips); the MySQL tests cover both stored-row aliases
and aliases within one request.
##########
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:
Fixed in
https://github.com/apache/gravitino/commit/036f5077f2736114adbba548a1f29705524a0f38.
The Javadoc now explicitly describes an intentionally broader accepted code
set: PostgreSQL READ COMMITTED deadlocks use 40P01; 40001 also covers
stricter-isolation serialization failures and H2 deadlocks; 55P03 covers a
server-configured lock_timeout. It also explains the vendor timeout codes.
Classification behavior is unchanged, and the PR description is being aligned
with this explanation.
--
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]
