Copilot commented on code in PR #13487:
URL: https://github.com/apache/gravitino/pull/13487#discussion_r4216898556
##########
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 scoping issue as the versioned update: the versioned soft-delete
predicate should also include `metalake_id` to prevent cross-metalake
modification if IDs aren’t globally unique (or to be resilient to unusual ID
reuse). Add `AND metalake_id = #{previous.metalakeId}`.
##########
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:
The CAS update predicate doesn’t scope by `metalake_id`. If `statistic_id`
or `metadata_object_id` are not globally unique across metalakes (or if there’s
ever an ID-collision/restore scenario), this could update a row in the wrong
metalake. Add `AND metalake_id = #{previous.metalakeId}` (and ensure `previous`
carries it) for correctness and consistency with the select queries, which
already filter by metalake.
##########
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:
`batchInsertStatisticPOsOnDuplicateKeyUpdate` doesn’t appear to guard
against an empty input batch. If `pos` is empty, `listStatisticPOsByNames` will
render an `IN ()` clause (invalid SQL), and the method will still take
locks/transactions unnecessarily. Add an early-return when
`statisticEntities`/`pos` is empty (similar to `batchDeleteStatisticPOs`).
##########
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:
The JavaDoc says this query avoids loading “values or audit fields”, but it
composes `listStatisticPOsByEntityId(...)`, which (per its name/typical
implementation) selects full rows (including value and audit info). Either
update the comment to reflect that it filters to requested names (but still
selects full rows), or change the underlying select to project only the
required columns for snapshot/version checks.
##########
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:
This change turns what used to be a single batch upsert into per-statistic
insert/update calls, which can significantly increase DB round-trips for larger
batches. If large batches are expected, consider using a batched
executor/statement strategy (e.g., MyBatis batch executor) or a provider-level
multi-row SQL approach while still keeping the per-row version predicates, to
reduce latency and lock hold time.
--
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]
