jerryshao commented on code in PR #13487:
URL: https://github.com/apache/gravitino/pull/13487#discussion_r4230317611
##########
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:
Will throwing an exception lead to a half-updated result?
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java:
##########
@@ -100,11 +174,84 @@ 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;
+ }
+ int[] deleted = new int[] {0};
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ LiveEndpointService.lockLiveEndpoint(identifier, type, observed);
+ for (String name : orderedNames) {
+ StatisticPO old = previous.get(name);
+ if (old == null) {
+ continue;
+ }
+ int updated =
+ SessionUtils.getWithoutCommit(
+ StatisticMetaMapper.class, mapper ->
mapper.deleteStatisticPOWithVersion(old));
+ if (updated == 1) {
+ deleted[0]++;
+ } else if (hasLiveStatistic(observed, name)) {
+ throw statisticConflict(null, name, identifier);
+ }
+ // Otherwise a concurrent drop already removed it. Like a drop of
a missing name, that
+ // is not a conflict and is simply not counted.
+ }
+ });
+ return deleted[0];
Review Comment:
The same question also applies to here.
##########
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));
Review Comment:
How many db access do we need typically? Can you use one access?
##########
core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/StatisticBaseSQLProvider.java:
##########
@@ -33,47 +33,42 @@
public class StatisticBaseSQLProvider {
- public String batchInsertStatisticPOsOnDuplicateKeyUpdate(
- @Param("statisticPOs") List<StatisticPO> statisticPOs) {
- return "<script>"
- + "INSERT INTO "
+ /** Inserts a new live statistic without upsert fallback. */
+ public String insertStatisticPO(@Param("statisticPO") StatisticPO
statisticPO) {
+ return "INSERT INTO "
+ STATISTIC_META_TABLE_NAME
+ " (statistic_id, statistic_name, statistic_value, metalake_id,
metadata_object_id,"
- + " metadata_object_type, audit_info, current_version, last_version,
deleted_at) VALUES "
- + "<foreach collection='statisticPOs' item='item' separator=','>"
- + "(#{item.statisticId}, "
- + "#{item.statisticName}, "
- + "#{item.statisticValue}, "
- + "#{item.metalakeId}, "
- + "#{item.metadataObjectId}, "
- + "#{item.metadataObjectType}, "
- + "#{item.auditInfo}, "
- + "#{item.currentVersion}, "
- + "#{item.lastVersion}, "
- + "#{item.deletedAt})"
- + "</foreach>"
- + " ON DUPLICATE KEY UPDATE "
- + " statistic_value = VALUES(statistic_value),"
- + " audit_info = VALUES(audit_info),"
- + " current_version = VALUES(current_version),"
- + " last_version = VALUES(last_version),"
- + " deleted_at = VALUES(deleted_at)"
- + "</script>";
+ + " metadata_object_type, audit_info, current_version, last_version,
deleted_at) VALUES"
+ + " (#{statisticPO.statisticId}, #{statisticPO.statisticName},"
+ + " #{statisticPO.statisticValue}, #{statisticPO.metalakeId},"
+ + " #{statisticPO.metadataObjectId},
#{statisticPO.metadataObjectType},"
+ + " #{statisticPO.auditInfo}, 1, 1, 0)";
Review Comment:
Don't hardcode the number here.
--
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]
