yuqi1129 commented on code in PR #13487:
URL: https://github.com/apache/gravitino/pull/13487#discussion_r4230519799
##########
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:
Fixed in 300e112df2. The insert (now `batchInsertStatisticPOs`) takes
`current_version`, `last_version` and `deleted_at` from the PO.
`StatisticPO.initializeStatisticPOs` already sets them to `INIT_VERSION` and
`DEFAULT_DELETED_AT`.
##########
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:
Fixed in 300e112df2. The per-statistic loop is gone. Inside the fenced
transaction, a write now runs at most two statements no matter how many
statistics are in the batch: one multi-row `INSERT` for new names and one
version-checked `UPDATE ... SET statistic_value = CASE statistic_id ...` for
existing ones. A drop runs one version-checked soft-delete `UPDATE`, and reads
the live rows again only when the count falls short, to tell a conflict from a
concurrent drop. The name lookup before the transaction and the
`lockLiveEndpoint` fence are unchanged. We can't merge everything into one
statement, because the CAS needs the versions observed before the write and the
fence has to come first. Conflicts are detected from the total row count, so
the old reason for separate statements (batch executors may return
`SUCCESS_NO_INFO`) no longer applies.
`testBatchReplacementKeepsValuesPerStatistic` checks that each row gets its own
value, on H2, MySQL and PostgreSQL.
##########
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:
No. The statements and the conflict checks all run inside one
`SessionUtils.doMultipleWithCommit`, and it rolls back on any `Throwable`
before rethrowing. A conflict therefore leaves nothing behind: an insert that
already ran is undone when the update in the same batch conflicts.
`testStatisticWriteConflictRollsBackWholeBatch` covers this: a mixed batch with
a stale update leaves both the existing statistic and the new insert unchanged.
It passes on H2, MySQL and PostgreSQL in 300e112df2.
##########
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:
Same as for the write path: the batch soft delete and the conflict check run
in one `doMultipleWithCommit`, so a conflict rolls back rows already deleted by
the same statement. `testStatisticDeleteConflictRollsBackWholeBatch` verifies
that a deleted `a_existing` is restored when `test` conflicts. The new
`testBatchDropCountsOnlyItsOwnDeletes` covers the case where one name was
already dropped concurrently: the call succeeds and counts only its own delete.
--
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]
