jerryshao commented on code in PR #13487:
URL: https://github.com/apache/gravitino/pull/13487#discussion_r4227838323
##########
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)";
Review Comment:
[Nit] The literals `1, 1, 0` match `POConverters.INIT_VERSION` and
`POConverters.DEFAULT_DELETED_AT` (POConverters.java:101-102), which
`StatisticPO.initializeStatisticPOs` already sets on every PO
(StatisticPO.java:106-108). Fixing them in SQL is the safer choice for a strict
insert, but it means the mapper accepts a full `StatisticPO` and silently
ignores three of its fields. Worth one line on
`StatisticMetaMapper.insertStatisticPO` saying that version and `deleted_at`
are fixed by the statement, so a future caller does not expect its own values
to be honoured.
Verified by: both constants are 1 and 0 today, so current behaviour is
identical; this is about the mapper contract, not a behaviour change.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java:
##########
@@ -63,6 +77,20 @@ public List<StatisticEntity> listStatisticsByEntity(
.collect(Collectors.toList());
}
+ /**
+ * Creates or replaces statistics of a metadata object by name.
+ *
+ * <p>The name keeps the historical upsert name and metric, but the write is
no longer a blind
Review Comment:
[Nit] Since the write is no longer an upsert, the method name is worth
changing rather than explaining: `baseMetricName` is an explicit literal on the
`@Monitored` annotation just below (line 96), so a Java rename such as
`writeStatisticsWithVersion` keeps the emitted metric identical, and
`JDBCBackend.java:676` is the only call site to update. That removes the need
for this sentence and stops the next reader from assuming an upsert.
Verified by: the annotation sets `baseMetricName =
"batchInsertStatisticPOsOnDuplicateKeyUpdate"` independently of the method
name, and `grep` shows JDBCBackend as the sole caller.
##########
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}"
+ + " AND metadata_object_id = #{previous.metadataObjectId}"
+ + " AND current_version = #{previous.currentVersion} AND deleted_at =
0";
Review Comment:
[Nit] This predicate omits `statistic_name`, while
`updateStatisticPOWithVersion` includes it (line 58). `statistic_id` is the
primary key on PostgreSQL (scripts/postgresql/schema-2.0.0-postgresql.sql:772)
but only a non-unique secondary index on MySQL (`KEY idx_stid`,
scripts/mysql/schema-2.0.0-mysql.sql:441), so the two CAS statements rest on
slightly different assumptions for no reason. Adding `AND statistic_name =
#{previous.statisticName}` costs nothing and makes them symmetric.
Verified by: comparing both predicates in this file against the MySQL and
PostgreSQL schema definitions for `statistic_meta`.
##########
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:
[Important] This conflict is now reachable from Gravitino's own automated
statistic writer, which has no retry anywhere on the path.
`GravitinoStatisticsUpdater.doUpdateTableStatistics`
(maintenance/updaters/src/main/java/org/apache/gravitino/maintenance/optimizer/updater/statistics/GravitinoStatisticsUpdater.java:75-86)
calls the client's `updateStatistics` exactly once. The client turns the new
409 into `OptimisticLockException`
(clients/client-java/src/main/java/org/apache/gravitino/client/ErrorHandlers.java:1428),
and `IcebergUpdateStatsAndMetricsJob` catches every `Exception` with
`LOG.error` plus `System.exit(1)`
(maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java:134-137).
So an optimizer run that races a user update of the same statistic name used
to be a silent upsert and now aborts the whole Spark job.
The in-process tree lock in `StatisticManager.updateStatistics`
(core/src/main/java/org/apache/gravitino/stats/StatisticManager.java:203-210)
serializes writers inside one server, so this needs two servers, or a client
racing the optimizer - which is exactly the multi-writer case this PR is about.
Suggest either a bounded retry around the updater call or a follow-up issue
linked from the new docs paragraph, so the automated writer is not the one
caller that cannot honour the retry advice.
Verified by: reading both updater call sites and the client error handler on
this checkout; `grep -rn "OptimisticLockException" maintenance/` finds no
handling at all.
--
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]
