yuqi1129 commented on code in PR #13487:
URL: https://github.com/apache/gravitino/pull/13487#discussion_r4228086254
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java:
##########
@@ -79,19 +107,65 @@ 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()));
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ LiveEndpointService.lockLiveEndpoint(entity, type,
namespacedEntityId);
+ // Execute each CAS separately: a batch executor may report
SUCCESS_NO_INFO rather than
+ // the per-row counts needed to detect conflicts (for example, MySQL
batch rewriting).
+ for (StatisticPO po : pos) {
+ StatisticPO old = previous.get(po.getStatisticName());
+ int updated;
+ try {
+ updated =
+ SessionUtils.getWithoutCommit(
+ StatisticMetaMapper.class,
+ mapper ->
+ old == null
+ ? mapper.insertStatisticPO(po)
+ : mapper.updateStatisticPOWithVersion(po, old));
+ } catch (RuntimeException e) {
+ // A writer can create the same statistic after the snapshot
above. A duplicate
+ // insert is a stale snapshot, not an internal server error.
+ if (old == null && isDuplicateKey(e)) {
+ throw statisticConflict(e, po.getStatisticName(), entity);
+ }
+ throw e;
+ }
+ if (updated != 1) {
+ throw statisticConflict(null, po.getStatisticName(), entity);
+ }
Review Comment:
Removed the automatic retry in `138b3a3fd8`.
An OCC conflict is an expected outcome of concurrent writes: the whole batch
is rolled back and the caller receives `OptimisticLockException` / HTTP 409.
Recovery is an application decision. The client should handle the exception and
choose whether to abort or retry against the current state; this PR does not
impose an automatic retry policy.
The built-in statistics updater therefore keeps one write attempt and
propagates the exception. A maintenance job can fail on that conflict through
its existing error handling, which is accepted here. The Java API and
statistics documentation now state this explicitly, and a regression test
verifies that the same exception is propagated after exactly one write call.
--
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]
