yuqi1129 commented on code in PR #13487:
URL: https://github.com/apache/gravitino/pull/13487#discussion_r4219318768
##########
core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/StatisticBaseSQLProvider.java:
##########
@@ -33,47 +33,41 @@
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)";
}
- public String batchDeleteStatisticPOs(
- @Param("entityId") Long entityId, @Param("statisticNames") List<String>
statisticNames) {
- return "<script>"
- + "UPDATE "
+ /** Updates the value and advances the version when the observed row still
matches. */
+ public String updateStatisticPOWithVersion(
+ @Param("statisticPO") StatisticPO statisticPO, @Param("previous")
StatisticPO previous) {
+ return "UPDATE "
+ + STATISTIC_META_TABLE_NAME
+ + " SET statistic_value = #{statisticPO.statisticValue},"
+ + " audit_info = #{statisticPO.auditInfo},"
+ + " last_version = current_version, current_version = current_version
+ 1"
+ + " WHERE statistic_id = #{previous.statisticId}"
+ + " AND metadata_object_id = #{previous.metadataObjectId}"
+ + " AND statistic_name = #{previous.statisticName}"
+ + " AND current_version = #{previous.currentVersion} AND deleted_at =
0";
+ }
Review Comment:
`statistic_id` and `metadata_object_id` come from the global random ID
generator, so they are unique across metalakes, and the unique key
`(statistic_name, metadata_object_id, deleted_at)` has no metalake either. An
extra `metalake_id` predicate would not change which row matches.
##########
core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/StatisticBaseSQLProvider.java:
##########
@@ -33,47 +33,41 @@
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)";
}
- public String batchDeleteStatisticPOs(
- @Param("entityId") Long entityId, @Param("statisticNames") List<String>
statisticNames) {
- return "<script>"
- + "UPDATE "
+ /** Updates the value and advances the version when the observed row still
matches. */
+ public String updateStatisticPOWithVersion(
+ @Param("statisticPO") StatisticPO statisticPO, @Param("previous")
StatisticPO previous) {
+ return "UPDATE "
+ + STATISTIC_META_TABLE_NAME
+ + " SET statistic_value = #{statisticPO.statisticValue},"
+ + " audit_info = #{statisticPO.auditInfo},"
+ + " last_version = current_version, current_version = current_version
+ 1"
+ + " WHERE statistic_id = #{previous.statisticId}"
+ + " AND metadata_object_id = #{previous.metadataObjectId}"
+ + " AND statistic_name = #{previous.statisticName}"
+ + " AND current_version = #{previous.currentVersion} AND deleted_at =
0";
+ }
+
+ /** Soft-deletes the observed statistic and advances its version. */
+ public String deleteStatisticPOWithVersion(@Param("previous") StatisticPO
previous) {
+ return "UPDATE "
+ STATISTIC_META_TABLE_NAME
+ softDeleteSQL()
- + " WHERE "
- + " statistic_name IN ("
- + "<foreach collection='statisticNames' item='item' separator=','>"
- + " #{item}"
- + "</foreach>"
- + " ) AND deleted_at = 0 AND metadata_object_id = #{entityId}"
- + "</script>";
+ + ", last_version = current_version, current_version = current_version
+ 1"
+ + " WHERE statistic_id = #{previous.statisticId}"
Review Comment:
Same reasoning as the versioned update: the IDs are globally unique, so
`metalake_id` adds no scoping.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java:
##########
@@ -79,19 +106,61 @@ 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()));
Review Comment:
The method already returns early when `statisticEntities` is null or empty,
so `pos` is never empty and the `IN ()` case is unreachable.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java:
##########
@@ -79,19 +106,61 @@ 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);
+ 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:
Intentional: each row needs its own update count to detect a CAS conflict,
and batched execution on MySQL with `rewriteBatchedStatements` reports
`SUCCESS_NO_INFO` instead of per-row counts. Statistic batches are small, so
the extra round trips are acceptable.
##########
core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/StatisticBaseSQLProvider.java:
##########
@@ -93,6 +87,20 @@ public String listStatisticPOsByEntityId(
+ " WHERE metadata_object_id = #{entityId} AND deleted_at = 0 AND
metalake_id = #{metalakeId}";
}
+ /** Selects the requested live statistics without loading unrelated values
or audit fields. */
+ public String listStatisticPOsByNames(
+ @Param("metalakeId") Long metalakeId,
+ @Param("entityId") Long entityId,
+ @Param("names") List<String> names) {
+ return "<script>"
+ + listStatisticPOsByEntityId(metalakeId, entityId)
+ + " AND statistic_name IN "
+ + "<foreach collection='names' item='name' open='(' separator=','
close=')'>"
+ + "#{name}"
+ + "</foreach>"
+ + "</script>";
+ }
Review Comment:
Fixed in 183ea7207d: the Javadoc now says it selects full rows, limited to
the requested names.
--
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]
