This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 8bd56bd493 [#13177] fix(core): Add statistic CAS and live-target
checks (#13487)
8bd56bd493 is described below
commit 8bd56bd49308b7d79277a62e7899e8ce38177de7
Author: Qi Yu <[email protected]>
AuthorDate: Mon Oct 12 14:39:25 2026 +0800
[#13177] fix(core): Add statistic CAS and live-target checks (#13487)
### What changes were proposed in this pull request?
- Add version CAS to statistic updates and deletes, and fence statistic
writes with `lockLiveEndpoint` from #13004 (replacing the schema-only
write lock).
- Report concurrent first creation of the same statistic as an OCC
conflict.
- A drop that loses to a concurrent drop of the same statistic is not a
conflict; it is simply not counted, as before.
- Load only the requested statistic names for write snapshots.
- Add regression tests for batch rollback and same-name replacement
conflicts, and document OCC outcomes and caller-controlled recovery in
REST and Java API contracts.
- Rename the service write method to `writeStatisticsWithVersion` while
preserving its explicit metric name. Make update/delete identity
predicates symmetric.
- Write each batch with one statement per kind inside the fenced
transaction: one multi-row insert for new statistics, one
version-checked update for replacements, and one version-checked soft
delete for drops. Any row-count mismatch fails and rolls back the whole
batch. The insert takes versions and `deleted_at` from the PO instead of
hard-coded values.
- A batch statement locks its rows in the order of the database plan, so
concurrent batches on another node can deadlock on MySQL. Deadlocks,
serialization failures and lock wait timeouts anywhere in the fenced
transaction (SQLState 40001/40P01/55P03, MySQL error codes 1213 and
1205, H2 error code 50200) are reported as the same OCC conflict (409)
instead of a 500, on both write and drop paths. This accepted code set
intentionally covers more than the defaults: PostgreSQL READ COMMITTED
emits 40P01 for deadlocks; 40001 also covers stricter isolation levels
and H2 deadlocks, while 55P03 covers a server-configured PostgreSQL
lock_timeout.
### Why are the changes needed?
Statistic writes previously used an unconditional upsert after resolving
the target ID outside the transaction. A concurrent target deletion
could leave an orphan, and concurrent statistic updates could silently
overwrite each other. Sequential replacement by name remains supported.
Fix: #13177
### Does this PR introduce _any_ user-facing change?
Concurrent statistic writes can return an OCC conflict (HTTP 409), and
writes to a deleted or replaced target fail with not found. No API or
configuration fields change.
Statistic writes and deletes now take shared locks on the metalake,
catalog, schema and target rows (the same fence as tag/policy relations)
instead of only the schema row (H2 takes exclusive row locks for this
fence). While a schema drop or nested schema create holds its catalog
lock, statistic writes in that catalog wait for it.
On MySQL, whose collation ignores trailing spaces, writing `name ` when
`name` already exists is now rejected with HTTP 400
(`IllegalStatisticNameException`) instead of silently overwriting
`name`, and dropping `name ` no longer drops `name`; H2 and PostgreSQL
store two distinct statistics. The same applies within one request: on
MySQL, a request containing both `name` and `name ` is rejected with
HTTP 400 before any statement runs. Because both kinds of alias are
rejected up front, any duplicate key that reaches the insert comes from
a concurrent creator and is a 409. Rejecting trailing spaces at the REST
layer for every backend could be considered separately.
OCC is an expected conflict outcome, and the whole conflicting batch is
rolled back. Clients decide whether to abort or retry against the
current state according to their application semantics. The built-in
statistics updater makes one write attempt and propagates the exception;
a maintenance job can therefore fail under its existing error handling.
This PR leaves recovery policy to callers.
### How was this patch tested?
- CI follow-up head `a509c8028c`: merged current main and removed the
obsolete Ossie import verification introduced by #12627 after #13512
changed the REST call path. Full non-Docker server suite: 641 passed; H2
`TestStatisticMetaService`: 20 passed; server/core Spotless and diff
checks passed. Fresh-context full-diff review passed. The isolated main
repair (`9a749a52dd`) also passed all 640 server tests, including 73
Semantic Model REST cases.
Added tests for deadlock/serialization/lock-timeout/duplicate-key
classification on both write and drop paths (including a real H2
lock-wait timeout), trailing-space aliases on write (against an existing
row and within one request), exact-name drops on a padding collation,
statistic 409 mapping in REST, per-statistic values in one batch
replacement, a batch drop that partly loses to a concurrent drop,
concurrent first creation, stale update/delete versions (mapper and
service layer), a target dropped after the version snapshot (write and
delete), two concurrent drops of the same statistic, requested-name
snapshots, whole-batch rollback after a later conflict, and same-name
statistic recreation at version one.
- Current head `036f5077f2`: full non-Docker core suite 2,680 passed / 2
existing disabled tests; server 577 passed; optimizer 234 passed / 5
existing environment-IT cases skipped. Core Spotless and Javadoc checks
passed, and `:docs:build` passed with a writable temporary npm cache.
- Current head: the real H2/MySQL/PostgreSQL `TestStatisticMetaService`
suite ran 60 invocations (20 per backend): 58 passed, 2 skipped because
the real lock-timeout test is H2-only. Mutation verification removed all
three trailing-space normalizations: both MySQL alias cases failed their
intended assertions, while the four H2/PostgreSQL cases passed. The
mutation was restored in the isolated validation checkout.
- `212b8f151c`: `TestStatisticMetaService` 60 run on H2, MySQL and
PostgreSQL (58 passed, 2 skipped: the H2-only real lock-timeout test on
MySQL and PostgreSQL), `TestStatisticManager` 3 and server
`TestStatisticOperations` 17 passed, `:docs:build` passed. Removing H2
50200 from the conflict codes fails both the synthetic and the real H2
lock-timeout test.
- Earlier head: `TestStatisticMetaService` 57 passed on H2, MySQL and
PostgreSQL (19 each), `TestStatisticManager` 3,
`TestOrphanedMetadataObjectRelationService` 9, all passed; server
`TestStatisticOperations` 17 passed (server code unchanged since).
Mutation checks: taking the replacement value from the old row or
disabling the live-name conflict check failed 7 methods on all three
backends; skipping conflict translation on the drop path or matching
MySQL 1205 only through the SQLState failed the classification test on
all three backends; disabling the trailing-space alias checks failed
their tests on MySQL (the only padding collation).
- An independent review reproduced the plan-dependent deadlock on MySQL
8.0 with the same statement shape; the fix maps it to 409. It was not
reproduced end to end through Gravitino.
The updater regression asserts that an OCC exception is propagated
unchanged after exactly one write call. The mapper regression verifies
that a mismatched statistic name cannot delete a row even when other
observed fields match. No full Spark/client-server concurrent job run
was executed; existing environment ITs remain gated separately.
---
.../apache/gravitino/stats/SupportsStatistics.java | 16 +
.../gravitino/api/stats/supports_statistics.py | 14 +
.../gravitino/storage/relational/JDBCBackend.java | 2 +-
.../relational/mapper/StatisticMetaMapper.java | 44 +-
.../mapper/StatisticSQLProviderFactory.java | 44 +-
.../provider/base/StatisticBaseSQLProvider.java | 104 ++-
.../postgresql/StatisticPostgresSQLProvider.java | 31 -
.../relational/service/LiveEndpointService.java | 4 +-
.../relational/service/StatisticMetaService.java | 351 ++++++++--
.../service/TestStatisticMetaService.java | 778 ++++++++++++++++++++-
docs/manage-statistics-in-gravitino.md | 18 +
docs/open-api/statistics.yaml | 4 +-
.../statistics/TestGravitinoStatisticsUpdater.java | 26 +
.../web/rest/TestSemanticModelOperations.java | 1 -
.../server/web/rest/TestStatisticOperations.java | 47 ++
15 files changed, 1342 insertions(+), 142 deletions(-)
diff --git
a/api/src/main/java/org/apache/gravitino/stats/SupportsStatistics.java
b/api/src/main/java/org/apache/gravitino/stats/SupportsStatistics.java
index 5923b4a089..0f9bf890a6 100644
--- a/api/src/main/java/org/apache/gravitino/stats/SupportsStatistics.java
+++ b/api/src/main/java/org/apache/gravitino/stats/SupportsStatistics.java
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.Map;
import org.apache.gravitino.annotation.Evolving;
import org.apache.gravitino.exceptions.IllegalStatisticNameException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.exceptions.UnmodifiableStatisticException;
/**
@@ -44,7 +45,14 @@ public interface SupportsStatistics {
* unmodifiable, it will throw an UnmodifiableStatisticException. If the
statistic name is
* illegal, it will throw an IllegalStatisticNameException.
*
+ * <p>For Gravitino-managed table statistics, a concurrent creation,
modification or deletion can
+ * fail the entire update with {@link OptimisticLockException}. No part of
that update is
+ * committed. The caller decides whether to abort or retry against the
current state; Gravitino
+ * does not automatically retry the write.
+ *
* @param statistics a map of statistic names to their values
+ * @throws OptimisticLockException if a Gravitino-managed table statistic
was concurrently
+ * created, modified or deleted
*/
void updateStatistics(Map<String, StatisticValue<?>> statistics)
throws UnmodifiableStatisticException, IllegalStatisticNameException;
@@ -53,9 +61,17 @@ public interface SupportsStatistics {
* Drop statistics by their names. If the statistic is unmodifiable, it will
throw an
* UnmodifiableStatisticException.
*
+ * <p>For Gravitino-managed table statistics, a concurrent modification or
same-name replacement
+ * can fail the entire drop with {@link OptimisticLockException}. No part of
that drop is
+ * committed. The caller decides whether to abort or retry against the
current state; Gravitino
+ * does not automatically retry the write. A statistic already removed by
another drop is not a
+ * conflict and is not counted as dropped by this call.
+ *
* @param statistics a list of statistic names to be dropped
* @return true if the statistics were successfully dropped, false if no
statistics were dropped
* @throws UnmodifiableStatisticException if any of the statistics to be
dropped are unmodifiable
+ * @throws OptimisticLockException if a Gravitino-managed table statistic
was concurrently
+ * modified or replaced
*/
boolean dropStatistics(List<String> statistics) throws
UnmodifiableStatisticException;
}
diff --git a/clients/client-python/gravitino/api/stats/supports_statistics.py
b/clients/client-python/gravitino/api/stats/supports_statistics.py
index a0fcc28a02..c389a0be73 100644
--- a/clients/client-python/gravitino/api/stats/supports_statistics.py
+++ b/clients/client-python/gravitino/api/stats/supports_statistics.py
@@ -49,9 +49,16 @@ class SupportsStatistics(ABC):
Args:
statistics: a map of statistic names to their values
+ For Gravitino-managed table statistics, a concurrent creation,
modification or deletion
+ can fail the entire update with `OptimisticLockException`. No part of
that update is
+ committed. The caller decides whether to abort or retry against the
current state;
+ Gravitino does not automatically retry the write.
+
Raises:
IllegalStatisticNameException: If the statistic name is illegal
UnmodifiableStatisticException: If the statistic is unmodifiable
+ OptimisticLockException: If a Gravitino-managed table statistic
was concurrently
+ created, modified or deleted
"""
@abstractmethod
@@ -68,7 +75,14 @@ class SupportsStatistics(ABC):
`True` if the statistics were successfully dropped,
`False` if no statistics were dropped
+ For Gravitino-managed table statistics, a concurrent modification or
same-name
+ replacement can fail the entire drop with `OptimisticLockException`.
No part of that drop
+ is committed. A statistic already removed by another drop is not a
conflict and is not
+ counted as dropped by this call.
+
Raises:
UnmodifiableStatisticException:
if any of the statistics to be dropped are unmodifiable
+ OptimisticLockException:
+ if a Gravitino-managed table statistic was concurrently
modified or replaced
"""
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
b/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
index 995b385a4e..096bbdadd8 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
@@ -671,7 +671,7 @@ public class JDBCBackend implements RelationalBackend,
SupportsOrphanedRelationC
"All entities must be in the same namespace for batchPut
operation.");
StatisticMetaService.getInstance()
- .batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ .writeStatisticsWithVersion(
statisticEntities,
NameIdentifier.parse(statisticEntities.get(0).namespace().toString()),
Entity.EntityType.TABLE);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/StatisticMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/StatisticMetaMapper.java
index 7c25383bed..df050e8e0f 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/StatisticMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/StatisticMetaMapper.java
@@ -34,15 +34,45 @@ public interface StatisticMetaMapper {
List<StatisticPO> listStatisticPOsByEntityId(
@Param("metalakeId") Long metalakeId, @Param("entityId") long entityId);
- @InsertProvider(
+ /** Lists only the named live statistics for a metadata object. */
+ @SelectProvider(type = StatisticSQLProviderFactory.class, method =
"listStatisticPOsByNames")
+ List<StatisticPO> listStatisticPOsByNames(
+ @Param("metalakeId") Long metalakeId,
+ @Param("entityId") long entityId,
+ @Param("names") List<String> names);
+
+ /**
+ * Inserts statistics in one statement. The whole statement fails with a
unique-key violation if
+ * any of them already has a live row with the same name and target.
+ *
+ * @param statisticPOs the statistics to insert, with their initial versions
and deleted_at
+ * @return the number of inserted rows
+ */
+ @InsertProvider(type = StatisticSQLProviderFactory.class, method =
"batchInsertStatisticPOs")
+ Integer batchInsertStatisticPOs(@Param("statisticPOs") List<StatisticPO>
statisticPOs);
+
+ /**
+ * Replaces statistic values in one statement, each only if its observed
version is still current.
+ *
+ * @param statisticPOs one PO per observed row: its ID, target, name and
current version identify
+ * the row, and its value and audit info are the replacement
+ * @return the number of replaced rows; fewer than requested means some rows
changed meanwhile
+ */
+ @UpdateProvider(
type = StatisticSQLProviderFactory.class,
- method = "batchInsertStatisticPOsOnDuplicateKeyUpdate")
- void batchInsertStatisticPOsOnDuplicateKeyUpdate(
- @Param("statisticPOs") List<StatisticPO> statisticPOs);
+ method = "batchUpdateStatisticPOsWithVersion")
+ Integer batchUpdateStatisticPOsWithVersion(@Param("statisticPOs")
List<StatisticPO> statisticPOs);
- @UpdateProvider(type = StatisticSQLProviderFactory.class, method =
"batchDeleteStatisticPOs")
- Integer batchDeleteStatisticPOs(
- @Param("entityId") Long entityId, @Param("statisticNames") List<String>
statisticNames);
+ /**
+ * Soft-deletes statistics in one statement, each only if its observed
version is still current.
+ *
+ * @param statisticPOs the observed rows, identified by ID, target, name and
current version
+ * @return the number of deleted rows; fewer than requested means some rows
changed meanwhile
+ */
+ @UpdateProvider(
+ type = StatisticSQLProviderFactory.class,
+ method = "batchDeleteStatisticPOsWithVersion")
+ Integer batchDeleteStatisticPOsWithVersion(@Param("statisticPOs")
List<StatisticPO> statisticPOs);
@UpdateProvider(
type = StatisticSQLProviderFactory.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/StatisticSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/StatisticSQLProviderFactory.java
index c15dc7ec98..c9608b8b1c 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/StatisticSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/StatisticSQLProviderFactory.java
@@ -31,7 +31,13 @@ import org.apache.ibatis.annotations.Param;
public class StatisticSQLProviderFactory {
- static class StatisticMySQLProvider extends StatisticBaseSQLProvider {}
+ static class StatisticMySQLProvider extends StatisticBaseSQLProvider {
+ // The shipped MySQL schema uses a PAD SPACE collation for statistic_meta.
+ @Override
+ public boolean namesIgnoreTrailingSpaces() {
+ return true;
+ }
+ }
static class StatisticH2Provider extends StatisticBaseSQLProvider {}
@@ -56,14 +62,32 @@ public class StatisticSQLProviderFactory {
return STATISTIC_SQL_PROVIDERS.get(jdbcBackendType);
}
- public static String batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ /**
+ * Returns whether the current backend treats statistic names that differ
only in trailing spaces
+ * as the same statistic.
+ *
+ * @return {@code true} for backends whose statistic key ignores trailing
spaces
+ */
+ public static boolean namesIgnoreTrailingSpaces() {
+ return getProvider().namesIgnoreTrailingSpaces();
+ }
+
+ /** Returns SQL for a strict multi-row statistic insert. */
+ public static String batchInsertStatisticPOs(
@Param("statisticPOs") List<StatisticPO> statisticPOs) {
- return
getProvider().batchInsertStatisticPOsOnDuplicateKeyUpdate(statisticPOs);
+ return getProvider().batchInsertStatisticPOs(statisticPOs);
}
- public static String batchDeleteStatisticPOs(
- @Param("entityId") Long entityId, @Param("statisticNames") List<String>
statisticNames) {
- return getProvider().batchDeleteStatisticPOs(entityId, statisticNames);
+ /** Returns SQL for a version-checked batch statistic update. */
+ public static String batchUpdateStatisticPOsWithVersion(
+ @Param("statisticPOs") List<StatisticPO> statisticPOs) {
+ return getProvider().batchUpdateStatisticPOsWithVersion(statisticPOs);
+ }
+
+ /** Returns SQL for a version-checked batch statistic soft delete. */
+ public static String batchDeleteStatisticPOsWithVersion(
+ @Param("statisticPOs") List<StatisticPO> statisticPOs) {
+ return getProvider().batchDeleteStatisticPOsWithVersion(statisticPOs);
}
public static String softDeleteStatisticsByEntityId(@Param("entityId") Long
entityId) {
@@ -75,6 +99,14 @@ public class StatisticSQLProviderFactory {
return getProvider().listStatisticPOsByEntityId(metalakeId, entityId);
}
+ /** Returns SQL to select only the named live statistics. */
+ public static String listStatisticPOsByNames(
+ @Param("metalakeId") Long metalakeId,
+ @Param("entityId") Long entityId,
+ @Param("names") List<String> names) {
+ return getProvider().listStatisticPOsByNames(metalakeId, entityId, names);
+ }
+
public static String softDeleteStatisticsByMetalakeId(@Param("metalakeId")
Long metalakeId) {
return getProvider().softDeleteStatisticsByMetalakeId(metalakeId);
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/StatisticBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/StatisticBaseSQLProvider.java
index 8a0c3d5b67..dd20247b89 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/StatisticBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/StatisticBaseSQLProvider.java
@@ -33,49 +33,75 @@ import org.apache.ibatis.annotations.Param;
public class StatisticBaseSQLProvider {
- public String batchInsertStatisticPOsOnDuplicateKeyUpdate(
- @Param("statisticPOs") List<StatisticPO> statisticPOs) {
- return "<script>"
- + "INSERT INTO "
+ /**
+ * Returns whether the backend's {@code statistic_meta} key ignores trailing
spaces, so that
+ * "name" and "name " are the same statistic.
+ *
+ * @return {@code true} if names that differ only in trailing spaces collide
+ */
+ public boolean namesIgnoreTrailingSpaces() {
+ return false;
+ }
+
+ /** Inserts new live statistics in one statement without upsert fallback. */
+ public String batchInsertStatisticPOs(@Param("statisticPOs")
List<StatisticPO> statisticPOs) {
+ return "<script>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 "
+ + " 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)"
+ + " (#{item.statisticId}, #{item.statisticName},
#{item.statisticValue},"
+ + " #{item.metalakeId}, #{item.metadataObjectId},
#{item.metadataObjectType},"
+ + " #{item.auditInfo}, #{item.currentVersion}, #{item.lastVersion},
#{item.deletedAt})"
+ + "</foreach></script>";
+ }
+
+ /**
+ * Replaces the values of the observed statistics in one statement and
advances their versions.
+ * Each PO identifies an observed row by ID, target, name and current
version, and carries the new
+ * value and audit info.
+ */
+ public String batchUpdateStatisticPOsWithVersion(
+ @Param("statisticPOs") List<StatisticPO> statisticPOs) {
+ return "<script>UPDATE "
+ + STATISTIC_META_TABLE_NAME
+ + " SET statistic_value = CASE statistic_id"
+ + "<foreach collection='statisticPOs' item='item'>"
+ + " WHEN #{item.statisticId} THEN #{item.statisticValue}"
+ + "</foreach> END,"
+ + " audit_info = CASE statistic_id"
+ + "<foreach collection='statisticPOs' item='item'>"
+ + " WHEN #{item.statisticId} THEN #{item.auditInfo}"
+ + "</foreach> END,"
+ + " last_version = current_version, current_version = current_version
+ 1"
+ + observedRowsPredicate()
+ "</script>";
}
- public String batchDeleteStatisticPOs(
- @Param("entityId") Long entityId, @Param("statisticNames") List<String>
statisticNames) {
- return "<script>"
- + "UPDATE "
+ /**
+ * Soft-deletes the observed statistics in one statement and advances their
versions. Each PO
+ * identifies an observed row by ID, target, name and current version.
+ */
+ public String batchDeleteStatisticPOsWithVersion(
+ @Param("statisticPOs") List<StatisticPO> statisticPOs) {
+ return "<script>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}"
+ + ", last_version = current_version, current_version = current_version
+ 1"
+ + observedRowsPredicate()
+ "</script>";
}
+ private static String observedRowsPredicate() {
+ return " WHERE deleted_at = 0 AND ("
+ + "<foreach collection='statisticPOs' item='item' separator=' OR '>"
+ + "(statistic_id = #{item.statisticId}"
+ + " AND metadata_object_id = #{item.metadataObjectId}"
+ + " AND statistic_name = #{item.statisticName}"
+ + " AND current_version = #{item.currentVersion})"
+ + "</foreach>)";
+ }
+
public String softDeleteStatisticsByEntityId(@Param("entityId") Long
entityId) {
return "UPDATE "
+ STATISTIC_META_TABLE_NAME
@@ -93,6 +119,20 @@ public class StatisticBaseSQLProvider {
+ " WHERE metadata_object_id = #{entityId} AND deleted_at = 0 AND
metalake_id = #{metalakeId}";
}
+ /** Selects full rows of only the requested live statistics of one metadata
object. */
+ 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>";
+ }
+
public String softDeleteStatisticsByMetalakeId(@Param("metalakeId") Long
metalakeId) {
return "UPDATE "
+ STATISTIC_META_TABLE_NAME
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/StatisticPostgresSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/StatisticPostgresSQLProvider.java
index ed6c5de7b7..de06021828 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/StatisticPostgresSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/StatisticPostgresSQLProvider.java
@@ -20,10 +20,8 @@ package
org.apache.gravitino.storage.relational.mapper.provider.postgresql;
import static
org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper.STATISTIC_META_TABLE_NAME;
-import java.util.List;
import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
import
org.apache.gravitino.storage.relational.mapper.provider.base.StatisticBaseSQLProvider;
-import org.apache.gravitino.storage.relational.po.StatisticPO;
public class StatisticPostgresSQLProvider extends StatisticBaseSQLProvider {
@Override
@@ -31,35 +29,6 @@ public class StatisticPostgresSQLProvider extends
StatisticBaseSQLProvider {
return " SET deleted_at = " + DatabaseTimeSQL.POSTGRESQL;
}
- @Override
- public String batchInsertStatisticPOsOnDuplicateKeyUpdate(List<StatisticPO>
statisticPOs) {
- return "<script>"
- + "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 CONFLICT (statistic_name, metadata_object_id, deleted_at)"
- + " DO UPDATE SET "
- + " statistic_value = EXCLUDED.statistic_value,"
- + " audit_info = EXCLUDED.audit_info,"
- + " current_version = EXCLUDED.current_version,"
- + " last_version = EXCLUDED.last_version,"
- + " deleted_at = EXCLUDED.deleted_at"
- + "</script>";
- }
-
@Override
public String deleteStatisticsByLegacyTimeline(Long legacyTimeline, int
limit) {
return "DELETE FROM "
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/LiveEndpointService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/LiveEndpointService.java
index 9f90b60f59..2b7d1ad9a9 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/LiveEndpointService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/LiveEndpointService.java
@@ -33,8 +33,8 @@ import org.apache.ibatis.session.SqlSession;
/**
* Fences metadata-object endpoints while a relation write is in progress.
*
- * <p>Statistic writers take a schema lock of their own today and adopt this
fence in #13177, so
- * nothing here covers them yet.
+ * <p>Statistic writes and deletes also take this fence, so a statistic cannot
be committed against
+ * a target that a concurrent drop has already removed.
*/
public final class LiveEndpointService {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java
index 2a429efb8a..7636af1356 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java
@@ -20,14 +20,31 @@ package org.apache.gravitino.storage.relational.service;
import static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME;
+import com.google.common.annotations.VisibleForTesting;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.Entity;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.exceptions.IllegalStatisticNameException;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.meta.NamespacedEntityId;
import org.apache.gravitino.meta.StatisticEntity;
import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
+import
org.apache.gravitino.storage.relational.mapper.StatisticSQLProviderFactory;
import org.apache.gravitino.storage.relational.po.StatisticPO;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.apache.gravitino.utils.NameIdentifierUtil;
@@ -38,13 +55,18 @@ import org.apache.gravitino.utils.NameIdentifierUtil;
*/
public class StatisticMetaService {
+ private static final int MYSQL_LOCK_WAIT_TIMEOUT = 1205;
+ private static final int MYSQL_DEADLOCK = 1213;
+ private static final int H2_LOCK_TIMEOUT = 50200;
+
private static final StatisticMetaService INSTANCE = new
StatisticMetaService();
public static StatisticMetaService getInstance() {
return INSTANCE;
}
- private StatisticMetaService() {}
+ @VisibleForTesting
+ StatisticMetaService() {}
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
@@ -63,10 +85,24 @@ public class StatisticMetaService {
.collect(Collectors.toList());
}
+ /**
+ * Creates or replaces statistics of a metadata object by name.
+ *
+ * <p>Existing statistics are replaced only if their version is unchanged
since this call read
+ * them, and missing statistics are inserted only if nobody created them
meanwhile; both cases
+ * otherwise fail the whole batch with {@link OptimisticLockException}. The
target is fenced in
+ * the same transaction, so a target dropped or replaced after its ID was
resolved fails with
+ * {@link NoSuchEntityException}.
+ *
+ * @param statisticEntities the statistics to write; names must be unique in
the batch
+ * @param entity the metadata object that owns the statistics
+ * @param type the metadata object type
+ */
+ // Preserve the historical metric name for compatibility with existing
monitoring.
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "batchInsertStatisticPOsOnDuplicateKeyUpdate")
- public void batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ public void writeStatisticsWithVersion(
List<StatisticEntity> statisticEntities, NameIdentifier entity,
Entity.EntityType type) {
if (statisticEntities == null || statisticEntities.isEmpty()) {
return;
@@ -79,19 +115,86 @@ public class StatisticMetaService {
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));
+ if (StatisticSQLProviderFactory.namesIgnoreTrailingSpaces()) {
+ rejectTrailingSpaceAliases(pos, entity);
+ }
+ Map<String, StatisticPO> previous =
+ listStatisticPOs(
+ namespacedEntityId,
+
pos.stream().map(StatisticPO::getStatisticName).collect(Collectors.toList()))
+ .stream()
+ .collect(Collectors.toMap(StatisticPO::getStatisticName,
Function.identity()));
+ // A collation that ignores trailing spaces (MySQL) returns the row of
"name" for "name ", and
+ // its unique key treats them as the same statistic. Writing "name " can
never succeed there, so
+ // it is rejected as an illegal name instead of a conflict that suggests a
retry.
+ for (String stored : previous.keySet()) {
+ if (!names.contains(stored)) {
+ String requested =
+ names.stream()
+ .filter(
+ name ->
+ StringUtils.stripEnd(name, "
").equals(StringUtils.stripEnd(stored, " ")))
+ .findFirst()
+ .orElse(stored);
+ throw new IllegalStatisticNameException(
+ "Statistic name '%s' is equivalent to the existing statistic '%s'
of %s in the backend"
+ + " collation; names must not differ only in trailing spaces",
+ requested, stored, entity);
+ }
+ }
+ List<StatisticPO> inserts = new ArrayList<>();
+ List<StatisticPO> updates = new ArrayList<>();
+ for (StatisticPO po : pos) {
+ StatisticPO old = previous.get(po.getStatisticName());
+ if (old == null) {
+ inserts.add(po);
+ } else {
+ updates.add(replacementOf(old, po));
+ }
+ }
+ // Each kind of write is one statement. Any mismatch fails the whole
transaction, so the batch
+ // either applies completely or not at all. Trailing-space aliases were
rejected above, so a
+ // duplicate key on the insert can only come from a concurrent creator and
is a conflict.
+ runFencedTransaction(
+ () -> {
+ LiveEndpointService.lockLiveEndpoint(entity, type,
namespacedEntityId);
+ if (!inserts.isEmpty()
+ && executeStatement(mapper ->
mapper.batchInsertStatisticPOs(inserts))
+ != inserts.size()) {
+ throw statisticConflict(null, names(inserts), entity);
+ }
+ if (!updates.isEmpty()
+ && executeStatement(mapper ->
mapper.batchUpdateStatisticPOsWithVersion(updates))
+ != updates.size()) {
+ throw statisticConflict(null, names(updates), entity);
+ }
+ },
+ pos,
+ entity);
}
+ /**
+ * Soft-deletes the named statistics of a metadata object.
+ *
+ * <p>Each statistic is deleted only at the version this call read. A
statistic that was changed
+ * meanwhile fails the whole batch with {@link OptimisticLockException}; one
that a concurrent
+ * drop already removed is not counted, like a name that does not exist. A
same-name replacement
+ * is a conflict even if its version matches the deleted row. A target
dropped or replaced after
+ * its ID was resolved fails with {@link NoSuchEntityException} when there
are rows to delete.
+ *
+ * @param identifier the metadata object that owns the statistics
+ * @param type the metadata object type
+ * @param statisticNames the statistic names to delete
+ * @return the number of statistics this call deleted
+ */
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "batchDeleteStatisticPOs")
@@ -100,11 +203,197 @@ public class StatisticMetaService {
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;
+ }
+ // Look up by the exact requested names. A collation that ignores trailing
spaces can return the
+ // row of "name" for a request of "name "; dropping "name " must not drop
"name".
+ List<StatisticPO> observedRows =
+ orderedNames.stream()
+ .map(previous::get)
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ if (observedRows.isEmpty()) {
+ return 0;
+ }
+ int[] deleted = new int[] {0};
+ runFencedTransaction(
+ () -> {
+ LiveEndpointService.lockLiveEndpoint(identifier, type, observed);
+ deleted[0] =
+ executeStatement(mapper ->
mapper.batchDeleteStatisticPOsWithVersion(observedRows));
+ if (deleted[0] == observedRows.size()) {
+ return;
+ }
+ // Some observed rows were not deleted. A name that is still live
was changed or
+ // replaced meanwhile, which is a conflict. Otherwise a concurrent
drop already removed
+ // it; like a drop of a missing name, that is not a conflict and is
simply not counted.
+ List<StatisticPO> live = liveStatistics(observed,
names(observedRows));
+ if (!live.isEmpty()) {
+ throw statisticConflict(null, names(live), identifier);
+ }
+ },
+ observedRows,
+ identifier);
+ return deleted[0];
+ }
- return SessionUtils.doWithCommitAndFetchResult(
+ @VisibleForTesting
+ List<StatisticPO> listStatisticPOs(NamespacedEntityId endpoint, List<String>
names) {
+ return SessionUtils.getWithoutCommit(
+ StatisticMetaMapper.class,
+ mapper ->
+ mapper.listStatisticPOsByNames(endpoint.namespaceIds()[0],
endpoint.entityId(), names));
+ }
+
+ /** Runs one batch statement in the current transaction and returns its row
count. */
+ @VisibleForTesting
+ int executeStatement(Function<StatisticMetaMapper, Integer> statement) {
+ return SessionUtils.getWithoutCommit(StatisticMetaMapper.class, statement);
+ }
+
+ /**
+ * Runs a fenced statistic transaction. A failure caused by a concurrent
writer anywhere in the
+ * transaction, including the fence's locking reads, is reported as a
conflict on {@code pos}.
+ */
+ private static void runFencedTransaction(
+ Runnable operations, List<StatisticPO> pos, NameIdentifier target) {
+ try {
+ SessionUtils.doMultipleWithCommit(operations);
+ } catch (RuntimeException e) {
+ throw translateConflict(e, pos, target);
+ }
+ }
+
+ /**
+ * Converts a failure caused by a concurrent writer into a conflict. A
duplicate insert means a
+ * statistic was created after the snapshot. A deadlock, serialization
failure or lock wait
+ * timeout can happen because a batch statement locks its rows in the order
of the database's
+ * plan, which differs between batches. The whole batch is rolled back
either way, so these are
+ * reported as a conflict rather than an internal server error.
+ */
+ private static RuntimeException translateConflict(
+ RuntimeException failure, List<StatisticPO> pos, NameIdentifier target) {
+ if (failure instanceof OptimisticLockException) {
+ return failure;
+ }
+ if (isDuplicateKey(failure)) {
+ return statisticConflict(failure, names(pos), target);
+ }
+ if (isConcurrencyFailure(failure)) {
+ return new OptimisticLockException(
+ failure,
+ "The statistics %s of %s could not be changed because of a
concurrent operation"
+ + " (deadlock or lock wait timeout); retry the operation",
+ names(pos),
+ target);
+ }
+ return failure;
+ }
+
+ /**
+ * Rejects requested names that the backend key treats as one statistic
because they differ only
+ * in trailing spaces. Such a request can never succeed, so it fails as an
illegal name before any
+ * write instead of as a conflict that suggests a retry.
+ */
+ private static void rejectTrailingSpaceAliases(List<StatisticPO> pos,
NameIdentifier target) {
+ Map<String, String> requestedByKey = new HashMap<>();
+ for (StatisticPO po : pos) {
+ String name = po.getStatisticName();
+ String other = requestedByKey.putIfAbsent(StringUtils.stripEnd(name, "
"), name);
+ if (other != null) {
+ throw new IllegalStatisticNameException(
+ "Statistic names '%s' and '%s' of %s are the same statistic in the
backend collation;"
+ + " names must not differ only in trailing spaces",
+ other, name, target);
+ }
+ }
+ }
+
+ private static OptimisticLockException statisticConflict(
+ @Nullable Throwable cause, List<String> names, NameIdentifier target) {
+ return new OptimisticLockException(
+ cause,
+ "One or more of the statistics %s of %s were changed concurrently;
retry the operation",
+ names,
+ target);
+ }
+
+ private static List<StatisticPO> liveStatistics(NamespacedEntityId endpoint,
List<String> names) {
+ return SessionUtils.getWithoutCommit(
StatisticMetaMapper.class,
- mapper -> mapper.batchDeleteStatisticPOs(entityId, statisticNames));
+ mapper ->
+ mapper.listStatisticPOsByNames(endpoint.namespaceIds()[0],
endpoint.entityId(), names));
+ }
+
+ private static List<String> names(List<StatisticPO> pos) {
+ return
pos.stream().map(StatisticPO::getStatisticName).sorted().collect(Collectors.toList());
+ }
+
+ /** Returns a PO that identifies the observed row and carries the
replacement value. */
+ private static StatisticPO replacementOf(StatisticPO observed, StatisticPO
replacement) {
+ return StatisticPO.builder()
+ .withStatisticId(observed.getStatisticId())
+ .withStatisticName(observed.getStatisticName())
+ .withMetalakeId(observed.getMetalakeId())
+ .withMetadataObjectId(observed.getMetadataObjectId())
+ .withMetadataObjectType(observed.getMetadataObjectType())
+ .withCurrentVersion(observed.getCurrentVersion())
+ .withLastVersion(observed.getLastVersion())
+ .withDeletedAt(observed.getDeletedAt())
+ .withStatisticValue(replacement.getStatisticValue())
+ .withAuditInfo(replacement.getAuditInfo())
+ .build();
+ }
+
+ /**
+ * Returns whether a failure is a unique-key violation. PostgreSQL and H2
report SQLState 23505,
+ * and MySQL reports error code 1062, the same codes the backend exception
converters match.
+ */
+ private static boolean isDuplicateKey(Throwable failure) {
+ return hasSqlException(failure, Set.of("23505"), Set.of(1062));
+ }
+
+ /**
+ * Returns whether a failure is a deadlock, serialization failure or lock
wait timeout caused by a
+ * concurrent writer. The accepted codes intentionally cover more than the
current default
+ * configuration emits. PostgreSQL reports 40P01 for deadlocks at our READ
COMMITTED isolation;
+ * 40001 also covers serialization failures at stricter isolation levels,
and 55P03 covers a
+ * server-configured lock_timeout. H2 reports 40001 for a deadlock and error
code 50200 (SQLState
+ * HYT00) for a lock wait timeout. MySQL reports error code 1213 for a
deadlock and 1205 for a
+ * lock wait timeout. Vendor error codes are matched because a timeout's
SQLState depends on the
+ * driver.
+ */
+ private static boolean isConcurrencyFailure(Throwable failure) {
+ return hasSqlException(
+ failure,
+ Set.of("40001", "40P01", "55P03"),
+ Set.of(MYSQL_DEADLOCK, MYSQL_LOCK_WAIT_TIMEOUT, H2_LOCK_TIMEOUT));
+ }
+
+ private static boolean hasSqlException(
+ Throwable failure, Set<String> sqlStates, Set<Integer> vendorErrorCodes)
{
+ for (Throwable cause = failure; cause != null; cause = cause.getCause()) {
+ if (cause instanceof SQLException) {
+ SQLException sql = (SQLException) cause;
+ if (sqlStates.contains(sql.getSQLState())
+ || vendorErrorCodes.contains(sql.getErrorCode())) {
+ return true;
+ }
+ }
+ }
+ return false;
}
@Monitored(
@@ -115,34 +404,4 @@ public class StatisticMetaService {
StatisticMetaMapper.class,
mapper -> mapper.deleteStatisticsByLegacyTimeline(legacyTimeline,
limit));
}
-
- private void doWithSchemaWriteLockIfNeeded(
- NameIdentifier identifier,
- Entity.EntityType type,
- NamespacedEntityId namespacedEntityId,
- Runnable writeOperation) {
- long[] namespaceIds = namespacedEntityId.namespaceIds();
- Long schemaId;
- switch (type) {
- case SCHEMA:
- schemaId = namespacedEntityId.entityId();
- break;
- case TABLE:
- case VIEW:
- case COLUMN:
- case FILESET:
- case TOPIC:
- case MODEL:
- case FUNCTION:
- schemaId = namespaceIds[2];
- break;
- default:
- SessionUtils.doMultipleWithCommit(writeOperation);
- return;
- }
-
- SchemaMetaService.getInstance()
- .doWithSchemaWriteLock(
- identifier, schemaId, namespaceIds[1], namespaceIds[0],
writeOperation);
- }
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestStatisticMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestStatisticMetaService.java
index a079abf5d8..014e4c7319 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestStatisticMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestStatisticMetaService.java
@@ -25,21 +25,29 @@ import java.sql.SQLException;
import java.sql.Statement;
import java.time.Instant;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.function.Function;
+import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
+import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
+import org.apache.gravitino.exceptions.IllegalStatisticNameException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.meta.CatalogEntity;
import org.apache.gravitino.meta.FilesetEntity;
import org.apache.gravitino.meta.ModelEntity;
+import org.apache.gravitino.meta.NamespacedEntityId;
import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.meta.StatisticEntity;
import org.apache.gravitino.meta.TableEntity;
@@ -49,16 +57,758 @@ import org.apache.gravitino.stats.StatisticValues;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
import org.apache.gravitino.storage.relational.po.SchemaPO;
+import org.apache.gravitino.storage.relational.po.StatisticPO;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.apache.ibatis.exceptions.PersistenceException;
import org.apache.ibatis.session.SqlSession;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.TestTemplate;
public class TestStatisticMetaService extends TestJDBCBackend {
private final StatisticMetaService statisticMetaService =
StatisticMetaService.getInstance();
+ /** Verifies the version snapshot query excludes unrelated statistic rows. */
+ @TestTemplate
+ public void testStatisticSnapshotLoadsOnlyRequestedNames() throws Exception {
+ String metalake = "statistic_filtered_metalake";
+ String catalog = "statistic_filtered_catalog";
+ String schema = "statistic_filtered_schema";
+ AuditInfo auditInfo =
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+ createParentEntities(metalake, catalog, schema, auditInfo);
+ TableEntity table =
+ createAndInsertTableEntity(Namespace.of(metalake, catalog, schema),
"statistic_filtered");
+ StatisticEntity unrelated =
+ TableStatisticEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("unrelated")
+ .withValue(StatisticValues.longValue(2L))
+ .withAuditInfo(auditInfo)
+ .build();
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createStatisticEntity(auditInfo, 1L), unrelated),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+
+ NamespacedEntityId endpoint =
+ EntityIdService.getEntityIds(table.nameIdentifier(),
Entity.EntityType.TABLE);
+ List<StatisticPO> selected =
statisticMetaService.listStatisticPOs(endpoint, List.of("test"));
+ Assertions.assertEquals(1, selected.size());
+ Assertions.assertEquals("test", selected.get(0).getStatisticName());
+ }
+
+ /** Verifies two writers that observe a missing statistic get one conflict
rather than a 500. */
+ @TestTemplate
+ public void testConcurrentFirstStatisticWritesReportConflict() throws
Exception {
+ String metalake = "statistic_first_write_metalake";
+ String catalog = "statistic_first_write_catalog";
+ String schema = "statistic_first_write_schema";
+ AuditInfo auditInfo =
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+ createParentEntities(metalake, catalog, schema, auditInfo);
+ TableEntity table =
+ createAndInsertTableEntity(
+ Namespace.of(metalake, catalog, schema), "statistic_first_write");
+
+ CyclicBarrier bothReadMissing = new CyclicBarrier(2);
+ StatisticMetaService racingService =
+ new StatisticMetaService() {
+ @Override
+ List<StatisticPO> listStatisticPOs(NamespacedEntityId endpoint,
List<String> names) {
+ List<StatisticPO> rows = super.listStatisticPOs(endpoint, names);
+ try {
+ bothReadMissing.await(30, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return rows;
+ }
+ };
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ Future<Throwable> first =
+ executor.submit(() -> writeFirstStatistic(racingService, table,
auditInfo, 1L));
+ Future<Throwable> second =
+ executor.submit(() -> writeFirstStatistic(racingService, table,
auditInfo, 2L));
+
+ Throwable firstFailure = first.get(30, TimeUnit.SECONDS);
+ Throwable secondFailure = second.get(30, TimeUnit.SECONDS);
+ Assertions.assertTrue((firstFailure == null) != (secondFailure == null));
+ Throwable conflict = firstFailure == null ? secondFailure : firstFailure;
+ Assertions.assertInstanceOf(OptimisticLockException.class, conflict);
+ // The duplicate-key failure stays attached for diagnosis.
+ Assertions.assertNotNull(conflict.getCause());
+ Assertions.assertEquals(
+ 1,
+ statisticMetaService
+ .listStatisticsByEntity(table.nameIdentifier(),
Entity.EntityType.TABLE)
+ .size());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ private Throwable writeFirstStatistic(
+ StatisticMetaService service, TableEntity table, AuditInfo auditInfo,
long value) {
+ try {
+ service.writeStatisticsWithVersion(
+ List.of(createStatisticEntity(auditInfo, value)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ }
+
+ /** Verifies stale mapper updates and deletes cannot change a newer
statistic version. */
+ @TestTemplate
+ public void testStatisticVersionCompareAndSet() throws Exception {
+ String metalake = "statistic_cas_metalake";
+ String catalog = "statistic_cas_catalog";
+ String schema = "statistic_cas_schema";
+ AuditInfo auditInfo =
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+ createParentEntities(metalake, catalog, schema, auditInfo);
+ TableEntity table =
+ createAndInsertTableEntity(Namespace.of(metalake, catalog, schema),
"statistic_cas_table");
+ StatisticEntity initial = createStatisticEntity(auditInfo, 10L);
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(initial), table.nameIdentifier(), Entity.EntityType.TABLE);
+ Long metalakeId =
+ EntityIdService.getEntityId(NameIdentifier.of(metalake),
Entity.EntityType.METALAKE);
+ StatisticPO stale =
+ SessionUtils.getWithoutCommit(
+ StatisticMetaMapper.class,
+ mapper -> mapper.listStatisticPOsByEntityId(metalakeId,
table.id()))
+ .get(0);
+ Assertions.assertEquals(1L, stale.getCurrentVersion());
+
+ // Identity, target and version alone must not permit a delete for a
different name.
+ StatisticPO wrongName =
+ StatisticPO.builder()
+ .withMetalakeId(stale.getMetalakeId())
+ .withStatisticId(stale.getStatisticId())
+ .withMetadataObjectId(stale.getMetadataObjectId())
+ .withMetadataObjectType(stale.getMetadataObjectType())
+ .withStatisticName("different_name")
+ .withStatisticValue(stale.getStatisticValue())
+ .withAuditInfo(stale.getAuditInfo())
+ .withCurrentVersion(stale.getCurrentVersion())
+ .withLastVersion(stale.getLastVersion())
+ .withDeletedAt(stale.getDeletedAt())
+ .build();
+ int wrongNameDeleted =
+ SessionUtils.getWithoutCommit(
+ StatisticMetaMapper.class,
+ mapper ->
mapper.batchDeleteStatisticPOsWithVersion(List.of(wrongName)));
+ Assertions.assertEquals(0, wrongNameDeleted);
+
+ StatisticEntity replacement = createStatisticEntity(auditInfo, 20L);
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(replacement), table.nameIdentifier(), Entity.EntityType.TABLE);
+ StatisticPO current =
+ SessionUtils.getWithoutCommit(
+ StatisticMetaMapper.class,
+ mapper -> mapper.listStatisticPOsByEntityId(metalakeId,
table.id()))
+ .get(0);
+ Assertions.assertEquals(stale.getStatisticId(), current.getStatisticId());
+ Assertions.assertEquals(2L, current.getCurrentVersion());
+ Assertions.assertEquals(1L, current.getLastVersion());
+ String staleValue =
+ StatisticPO.initializeStatisticPOs(
+ List.of(createStatisticEntity(auditInfo, 30L)),
+ metalakeId,
+ table.id(),
+ MetadataObject.Type.TABLE)
+ .get(0)
+ .getStatisticValue();
+ StatisticPO staleReplacement =
+ StatisticPO.builder()
+ .withMetalakeId(stale.getMetalakeId())
+ .withStatisticId(stale.getStatisticId())
+ .withMetadataObjectId(stale.getMetadataObjectId())
+ .withMetadataObjectType(stale.getMetadataObjectType())
+ .withStatisticName(stale.getStatisticName())
+ .withStatisticValue(staleValue)
+ .withAuditInfo(stale.getAuditInfo())
+ .withCurrentVersion(stale.getCurrentVersion())
+ .withLastVersion(stale.getLastVersion())
+ .withDeletedAt(stale.getDeletedAt())
+ .build();
+ int staleUpdated =
+ SessionUtils.getWithoutCommit(
+ StatisticMetaMapper.class,
+ mapper ->
mapper.batchUpdateStatisticPOsWithVersion(List.of(staleReplacement)));
+ Assertions.assertEquals(0, staleUpdated);
+ int staleDeleted =
+ SessionUtils.getWithoutCommit(
+ StatisticMetaMapper.class,
+ mapper ->
mapper.batchDeleteStatisticPOsWithVersion(List.of(stale)));
+ Assertions.assertEquals(0, staleDeleted);
+ Assertions.assertEquals(
+ 20L,
+ statisticMetaService
+ .listStatisticsByEntity(table.nameIdentifier(),
Entity.EntityType.TABLE)
+ .get(0)
+ .value()
+ .value());
+ }
+
+ /** Verifies a write or delete fails when its target is dropped after the
version snapshot. */
+ @TestTemplate
+ public void testStatisticWritesFailWhenTargetDroppedAfterSnapshot() throws
Exception {
+ String metalake = "statistic_dropped_target_metalake";
+ String catalog = "statistic_dropped_target_catalog";
+ String schema = "statistic_dropped_target_schema";
+ AuditInfo auditInfo =
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+ createParentEntities(metalake, catalog, schema, auditInfo);
+ Long metalakeId =
+ EntityIdService.getEntityId(NameIdentifier.of(metalake),
Entity.EntityType.METALAKE);
+
+ TableEntity writeTarget =
+ createAndInsertTableEntity(Namespace.of(metalake, catalog, schema),
"dropped_write");
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ dropTableAfterSnapshot(writeTarget)
+ .writeStatisticsWithVersion(
+ List.of(createStatisticEntity(auditInfo, 1L)),
+ writeTarget.nameIdentifier(),
+ Entity.EntityType.TABLE));
+ Assertions.assertEquals(0, countActiveStats(metalakeId));
+
+ TableEntity deleteTarget =
+ createAndInsertTableEntity(Namespace.of(metalake, catalog, schema),
"dropped_delete");
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createStatisticEntity(auditInfo, 1L)),
+ deleteTarget.nameIdentifier(),
+ Entity.EntityType.TABLE);
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ dropTableAfterSnapshot(deleteTarget)
+ .batchDeleteStatisticPOs(
+ deleteTarget.nameIdentifier(), Entity.EntityType.TABLE,
List.of("test")));
+ Assertions.assertEquals(0, countActiveStats(metalakeId));
+ }
+
+ /** Verifies a drop that loses to a concurrent drop of the same name is not
a conflict. */
+ @TestTemplate
+ public void testConcurrentDropOfSameStatisticIsNotConflict() throws
Exception {
+ String metalake = "statistic_double_drop_metalake";
+ String catalog = "statistic_double_drop_catalog";
+ String schema = "statistic_double_drop_schema";
+ AuditInfo auditInfo =
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+ createParentEntities(metalake, catalog, schema, auditInfo);
+ TableEntity table =
+ createAndInsertTableEntity(Namespace.of(metalake, catalog, schema),
"double_drop");
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createStatisticEntity(auditInfo, 1L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+
+ StatisticMetaService losingDrop =
+ new StatisticMetaService() {
+ @Override
+ List<StatisticPO> listStatisticPOs(NamespacedEntityId endpoint,
List<String> names) {
+ List<StatisticPO> rows = super.listStatisticPOs(endpoint, names);
+ Assertions.assertEquals(
+ 1,
+ statisticMetaService.batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE, names));
+ return rows;
+ }
+ };
+
+ Assertions.assertEquals(
+ 0,
+ losingDrop.batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE, List.of("test")));
+ Assertions.assertTrue(
+ statisticMetaService
+ .listStatisticsByEntity(table.nameIdentifier(),
Entity.EntityType.TABLE)
+ .isEmpty());
+ }
+
+ /** Verifies one batch statement replaces each statistic with its own value.
*/
+ @TestTemplate
+ public void testBatchReplacementKeepsValuesPerStatistic() throws Exception {
+ TableEntity table = createBatchConflictTable("replace");
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(
+ createNamedStatistic("a", 1L),
+ createNamedStatistic("b", 2L),
+ createNamedStatistic("c", 3L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(
+ createNamedStatistic("c", 30L),
+ createNamedStatistic("d", 40L),
+ createNamedStatistic("a", 10L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+
+ Map<String, StatisticEntity> current = statisticsByName(table);
+ Assertions.assertEquals(4, current.size());
+ Assertions.assertEquals(10L, current.get("a").value().value());
+ Assertions.assertEquals(2L, current.get("b").value().value());
+ Assertions.assertEquals(30L, current.get("c").value().value());
+ Assertions.assertEquals(40L, current.get("d").value().value());
+ }
+
+ /** Verifies a batch drop counts only its own deletes when a concurrent drop
removed one name. */
+ @TestTemplate
+ public void testBatchDropCountsOnlyItsOwnDeletes() throws Exception {
+ TableEntity table = createBatchConflictTable("partial_drop");
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createNamedStatistic("a", 1L), createNamedStatistic("b", 2L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+
+ StatisticMetaService losingDrop =
+ new StatisticMetaService() {
+ @Override
+ List<StatisticPO> listStatisticPOs(NamespacedEntityId endpoint,
List<String> names) {
+ List<StatisticPO> rows = super.listStatisticPOs(endpoint, names);
+ Assertions.assertEquals(
+ 1,
+ statisticMetaService.batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE,
List.of("a")));
+ return rows;
+ }
+ };
+
+ Assertions.assertEquals(
+ 1,
+ losingDrop.batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE, List.of("a",
"b")));
+ Assertions.assertTrue(statisticsByName(table).isEmpty());
+ }
+
+ /** Verifies concurrency failures of batch statements become conflicts on
writes and drops. */
+ @TestTemplate
+ public void testBatchStatementConcurrencyFailuresAreConflicts() throws
Exception {
+ TableEntity table = createBatchConflictTable("deadlock");
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createNamedStatistic("a", 1L)), table.nameIdentifier(),
Entity.EntityType.TABLE);
+
+ for (SQLException failure :
+ List.of(
+ new SQLException("MySQL deadlock", "40001", 1213),
+ new SQLException("PostgreSQL deadlock", "40P01"),
+ new SQLException("serialization failure", "40001"),
+ new SQLException("MySQL lock wait timeout", "40001", 1205),
+ new SQLException("MySQL lock wait timeout", "HY000", 1205),
+ new SQLException("H2 lock wait timeout", "HYT00", 50200),
+ new SQLException("PostgreSQL lock timeout", "55P03"))) {
+ StatisticMetaService failing = failingStatements(new
PersistenceException(failure));
+ OptimisticLockException writeConflict =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ failing.writeStatisticsWithVersion(
+ List.of(createNamedStatistic("a", 2L),
createNamedStatistic("b", 2L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE));
+ Assertions.assertSame(failure, writeConflict.getCause().getCause());
+ Assertions.assertTrue(writeConflict.getMessage().contains("[a, b]"));
+
Assertions.assertTrue(writeConflict.getMessage().contains(table.nameIdentifier().toString()));
+
+ OptimisticLockException dropConflict =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ failing.batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE,
List.of("a")));
+ Assertions.assertSame(failure, dropConflict.getCause().getCause());
+ }
+
+ // A duplicate key is a conflict on a write and on a drop. Names that
collide only in the
+ // backend
+ // collation are rejected before the write, so a duplicate key can only
come from a concurrent
+ // creator, even if that creator's row is gone again by the time the
caller looks.
+ SQLException duplicate = new SQLException("duplicate key", "23505");
+ StatisticMetaService duplicating = failingStatements(new
PersistenceException(duplicate));
+ OptimisticLockException writeDuplicate =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ duplicating.writeStatisticsWithVersion(
+ List.of(createNamedStatistic("b", 2L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE));
+ Assertions.assertSame(duplicate, writeDuplicate.getCause().getCause());
+ OptimisticLockException dropDuplicate =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ duplicating.batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE,
List.of("a")));
+ Assertions.assertSame(duplicate, dropDuplicate.getCause().getCause());
+
+ PersistenceException connectionFailure =
+ new PersistenceException(new SQLException("connection lost", "08006"));
+ StatisticMetaService failing = failingStatements(connectionFailure);
+ Assertions.assertSame(
+ connectionFailure,
+ Assertions.assertThrows(
+ PersistenceException.class,
+ () ->
+ failing.writeStatisticsWithVersion(
+ List.of(createNamedStatistic("a", 3L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE)));
+ Assertions.assertSame(
+ connectionFailure,
+ Assertions.assertThrows(
+ PersistenceException.class,
+ () ->
+ failing.batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE,
List.of("a"))));
+
+ Map<String, StatisticEntity> remaining = statisticsByName(table);
+ Assertions.assertEquals(1, remaining.size());
+ Assertions.assertEquals(1L, remaining.get("a").value().value());
+ }
+
+ /** Verifies a real H2 lock wait timeout on the fenced write is reported as
a conflict. */
+ @TestTemplate
+ public void testH2LockWaitTimeoutIsConflict() throws Exception {
+ Assumptions.assumeTrue("h2".equalsIgnoreCase(backendType), "H2 lock wait
timeout only");
+ TableEntity table = createBatchConflictTable("h2_lock_timeout");
+
+ // Hold the table row that the write fences on, so the write waits until
H2 gives up.
+ try (SqlSession holder =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(false);
+ Statement statement = holder.getConnection().createStatement()) {
+ try (ResultSet ignored =
+ statement.executeQuery(
+ "SELECT table_id FROM table_meta WHERE table_id = " + table.id()
+ " FOR UPDATE")) {
+ OptimisticLockException conflict =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createNamedStatistic("a", 1L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE));
+ Throwable cause = conflict;
+ while (cause != null && !(cause instanceof SQLException)) {
+ cause = cause.getCause();
+ }
+ Assertions.assertNotNull(cause, "the lock wait timeout stays attached
for diagnosis");
+ Assertions.assertEquals(50200, ((SQLException) cause).getErrorCode());
+ } finally {
+ holder.rollback();
+ }
+ }
+ Assertions.assertTrue(statisticsByName(table).isEmpty());
+ }
+
+ /** Verifies a write of a name equal to an existing one under a padding
collation is illegal. */
+ @TestTemplate
+ public void testWriteOfNameDifferingOnlyInTrailingSpaces() throws Exception {
+ TableEntity table = createBatchConflictTable("trailing_space_write");
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createNamedStatistic("name", 1L)), table.nameIdentifier(),
Entity.EntityType.TABLE);
+
+ List<StatisticEntity> padded = List.of(createNamedStatistic("name ", 2L));
+ if ("mysql".equalsIgnoreCase(backendType)) {
+ // MySQL's collation treats "name " as "name", so the write can never
succeed.
+ IllegalStatisticNameException e =
+ Assertions.assertThrows(
+ IllegalStatisticNameException.class,
+ () ->
+ statisticMetaService.writeStatisticsWithVersion(
+ padded, table.nameIdentifier(),
Entity.EntityType.TABLE));
+ // The message names both the requested name and the stored statistic it
collides with.
+ Assertions.assertTrue(e.getMessage().contains("'name '"),
e.getMessage());
+ Assertions.assertTrue(e.getMessage().contains("'name'"), e.getMessage());
+ Assertions.assertEquals(1, statisticsByName(table).size());
+ } else {
+ statisticMetaService.writeStatisticsWithVersion(
+ padded, table.nameIdentifier(), Entity.EntityType.TABLE);
+ Assertions.assertEquals(2L, statisticsByName(table).get("name
").value().value());
+ }
+ Assertions.assertEquals(1L,
statisticsByName(table).get("name").value().value());
+ }
+
+ /** Verifies names that collide only under a padding collation in one
request are illegal. */
+ @TestTemplate
+ public void testWriteOfTrailingSpaceAliasesInOneRequest() throws Exception {
+ TableEntity table = createBatchConflictTable("trailing_space_batch");
+ List<StatisticEntity> bothNew =
+ List.of(createNamedStatistic("x", 1L), createNamedStatistic("x ", 2L));
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createNamedStatistic("y", 1L)), table.nameIdentifier(),
Entity.EntityType.TABLE);
+ List<StatisticEntity> updateAndAlias =
+ List.of(createNamedStatistic("y", 10L), createNamedStatistic("y ",
20L));
+
+ if ("mysql".equalsIgnoreCase(backendType)) {
+ // MySQL's unique key treats "x " as "x", so these requests can never
succeed. They are
+ // rejected before any statement runs, so they cannot be mistaken for a
concurrent creator.
+ for (List<StatisticEntity> request : List.of(bothNew, updateAndAlias)) {
+ IllegalStatisticNameException e =
+ Assertions.assertThrows(
+ IllegalStatisticNameException.class,
+ () ->
+ statisticMetaService.writeStatisticsWithVersion(
+ request, table.nameIdentifier(),
Entity.EntityType.TABLE));
+ String alias = request.get(0).name();
+ Assertions.assertTrue(e.getMessage().contains("'" + alias + "'"),
e.getMessage());
+ Assertions.assertTrue(e.getMessage().contains("'" + alias + " '"),
e.getMessage());
+ }
+ Map<String, StatisticEntity> current = statisticsByName(table);
+ Assertions.assertEquals(1, current.size());
+ Assertions.assertEquals(1L, current.get("y").value().value());
+ } else {
+ statisticMetaService.writeStatisticsWithVersion(
+ bothNew, table.nameIdentifier(), Entity.EntityType.TABLE);
+ statisticMetaService.writeStatisticsWithVersion(
+ updateAndAlias, table.nameIdentifier(), Entity.EntityType.TABLE);
+ Map<String, StatisticEntity> current = statisticsByName(table);
+ Assertions.assertEquals(4, current.size());
+ Assertions.assertEquals(10L, current.get("y").value().value());
+ Assertions.assertEquals(20L, current.get("y ").value().value());
+ }
+ }
+
+ /**
+ * Verifies a drop matches only the exact requested name, even under a
padding collation. Only the
+ * MySQL backend, whose collation ignores trailing spaces, can return "name"
for "name ".
+ */
+ @TestTemplate
+ public void testDropDoesNotDropNameWithoutTrailingSpace() throws Exception {
+ TableEntity table = createBatchConflictTable("trailing_space");
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createNamedStatistic("name", 1L)), table.nameIdentifier(),
Entity.EntityType.TABLE);
+
+ Assertions.assertEquals(
+ 0,
+ statisticMetaService.batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE, List.of("name
")));
+ Assertions.assertEquals(1L,
statisticsByName(table).get("name").value().value());
+ }
+
+ /** Verifies a write or delete based on a stale version fails without
touching the newer value. */
+ @TestTemplate
+ public void testStaleStatisticWriteAndDeleteReportConflict() throws
Exception {
+ String metalake = "statistic_stale_service_metalake";
+ String catalog = "statistic_stale_service_catalog";
+ String schema = "statistic_stale_service_schema";
+ AuditInfo auditInfo =
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+ createParentEntities(metalake, catalog, schema, auditInfo);
+ TableEntity table =
+ createAndInsertTableEntity(Namespace.of(metalake, catalog, schema),
"stale_service");
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createStatisticEntity(auditInfo, 1L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+
+ OptimisticLockException staleWrite =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ updateAfterSnapshot(table, auditInfo, 2L)
+ .writeStatisticsWithVersion(
+ List.of(createStatisticEntity(auditInfo, 3L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE));
+ Assertions.assertTrue(staleWrite.getMessage().contains("retry the
operation"));
+ Assertions.assertEquals(2L, singleStatisticValue(table));
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ updateAfterSnapshot(table, auditInfo, 4L)
+ .batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE,
List.of("test")));
+ Assertions.assertEquals(4L, singleStatisticValue(table));
+ }
+
+ /** Verifies a late write conflict rolls back earlier updates and inserts in
the batch. */
+ @TestTemplate
+ public void testStatisticWriteConflictRollsBackWholeBatch() throws Exception
{
+ TableEntity table = createBatchConflictTable("write");
+ StatisticEntity unchanged = createNamedStatistic("a_existing", 10L);
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(unchanged, createStatisticEntity(AUDIT_INFO, 1L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+
+ OptimisticLockException conflict =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ updateAfterSnapshot(table, AUDIT_INFO, 2L)
+ .writeStatisticsWithVersion(
+ // A stale update in the batch must also roll back the
insert of b_new.
+ List.of(
+ createStatisticEntity(AUDIT_INFO, 3L),
+ createNamedStatistic("b_new", 30L),
+ createNamedStatistic("a_existing", 20L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE));
+ assertStatisticConflict(conflict, table);
+ Map<String, StatisticEntity> remaining = statisticsByName(table);
+ Assertions.assertEquals(2, remaining.size());
+ Assertions.assertEquals(unchanged.fields(),
remaining.get("a_existing").fields());
+ Assertions.assertEquals(2L, remaining.get("test").value().value());
+ }
+
+ /** Verifies a late delete conflict restores statistics deleted earlier in
the batch. */
+ @TestTemplate
+ public void testStatisticDeleteConflictRollsBackWholeBatch() throws
Exception {
+ TableEntity table = createBatchConflictTable("delete");
+ StatisticEntity unchanged = createNamedStatistic("a_existing", 10L);
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(unchanged, createStatisticEntity(AUDIT_INFO, 1L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+
+ OptimisticLockException conflict =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ updateAfterSnapshot(table, AUDIT_INFO, 2L)
+ .batchDeleteStatisticPOs(
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE,
+ List.of("test", "a_existing")));
+ assertStatisticConflict(conflict, table);
+ Map<String, StatisticEntity> remaining = statisticsByName(table);
+ Assertions.assertEquals(2, remaining.size());
+ Assertions.assertEquals(unchanged.fields(),
remaining.get("a_existing").fields());
+ Assertions.assertEquals(2L, remaining.get("test").value().value());
+ }
+
+ /** Verifies a stale write or delete cannot affect a same-name replacement
at version one. */
+ @TestTemplate
+ public void testRecreatedStatisticIsNotChangedByStaleWriteOrDelete() throws
Exception {
+ TableEntity table = createBatchConflictTable("recreated");
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createStatisticEntity(AUDIT_INFO, 1L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+ for (boolean delete : List.of(false, true)) {
+ StatisticEntity replacement = createStatisticEntity(AUDIT_INFO, 2L);
+ StatisticMetaService staleService =
+ new StatisticMetaService() {
+ @Override
+ List<StatisticPO> listStatisticPOs(NamespacedEntityId endpoint,
List<String> names) {
+ List<StatisticPO> rows = super.listStatisticPOs(endpoint, names);
+ Assertions.assertEquals(1L, rows.get(0).getCurrentVersion());
+ Assertions.assertEquals(
+ 1,
+ statisticMetaService.batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE, names));
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(replacement), table.nameIdentifier(),
Entity.EntityType.TABLE);
+ return rows;
+ }
+ };
+ OptimisticLockException conflict =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () -> {
+ if (delete) {
+ staleService.batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE,
List.of("test"));
+ } else {
+ staleService.writeStatisticsWithVersion(
+ List.of(createStatisticEntity(AUDIT_INFO, 3L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+ }
+ });
+ assertStatisticConflict(conflict, table);
+ Assertions.assertEquals(replacement.fields(),
statisticsByName(table).get("test").fields());
+ }
+ }
+
+ private TableEntity createBatchConflictTable(String suffix) throws Exception
{
+ String metalake = "batch_conflict_metalake_" + suffix;
+ String catalog = "batch_conflict_catalog";
+ String schema = "batch_conflict_schema";
+ createParentEntities(metalake, catalog, schema, AUDIT_INFO);
+ return createAndInsertTableEntity(Namespace.of(metalake, catalog, schema),
"batch_conflict");
+ }
+
+ private StatisticEntity createNamedStatistic(String name, long value) {
+ return TableStatisticEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(name)
+ .withValue(StatisticValues.longValue(value))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
+ private Map<String, StatisticEntity> statisticsByName(TableEntity table) {
+ return statisticMetaService
+ .listStatisticsByEntity(table.nameIdentifier(),
Entity.EntityType.TABLE)
+ .stream()
+ .collect(Collectors.toMap(StatisticEntity::name, statistic ->
statistic));
+ }
+
+ private void assertStatisticConflict(OptimisticLockException conflict,
TableEntity table) {
+ Assertions.assertTrue(conflict.getMessage().contains("test"));
+
Assertions.assertTrue(conflict.getMessage().contains(table.nameIdentifier().toString()));
+ Assertions.assertTrue(conflict.getMessage().contains("retry the
operation"));
+ }
+
+ private static StatisticMetaService failingStatements(RuntimeException
failure) {
+ return new StatisticMetaService() {
+ @Override
+ int executeStatement(Function<StatisticMetaMapper, Integer> statement) {
+ throw failure;
+ }
+ };
+ }
+
+ private StatisticMetaService updateAfterSnapshot(
+ TableEntity table, AuditInfo auditInfo, long value) {
+ return new StatisticMetaService() {
+ @Override
+ List<StatisticPO> listStatisticPOs(NamespacedEntityId endpoint,
List<String> names) {
+ List<StatisticPO> rows = super.listStatisticPOs(endpoint, names);
+ statisticMetaService.writeStatisticsWithVersion(
+ List.of(createStatisticEntity(auditInfo, value)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE);
+ return rows;
+ }
+ };
+ }
+
+ private Object singleStatisticValue(TableEntity table) {
+ List<StatisticEntity> statistics =
+ statisticMetaService.listStatisticsByEntity(
+ table.nameIdentifier(), Entity.EntityType.TABLE);
+ Assertions.assertEquals(1, statistics.size());
+ return statistics.get(0).value().value();
+ }
+
+ private StatisticMetaService dropTableAfterSnapshot(TableEntity table) {
+ return new StatisticMetaService() {
+ @Override
+ List<StatisticPO> listStatisticPOs(NamespacedEntityId endpoint,
List<String> names) {
+ List<StatisticPO> rows = super.listStatisticPOs(endpoint, names);
+
Assertions.assertTrue(TableMetaService.getInstance().deleteTable(table.nameIdentifier()));
+ return rows;
+ }
+ };
+ }
+
@TestTemplate
public void testTableStatisticWriteWaitsForConcurrentSchemaDelete() throws
Exception {
String metalakeName = "metalake_for_statistic_schema_fence";
@@ -257,7 +1007,7 @@ public class TestStatisticMetaService extends
TestJDBCBackend {
StatisticEntity statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, table.nameIdentifier(), Entity.EntityType.TABLE);
List<StatisticEntity> listEntities =
@@ -271,7 +1021,7 @@ public class TestStatisticMetaService extends
TestJDBCBackend {
statisticEntity = createStatisticEntity(auditInfo, 200L);
statisticEntities.clear();
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, table.nameIdentifier(), Entity.EntityType.TABLE);
listEntities =
@@ -346,25 +1096,25 @@ public class TestStatisticMetaService extends
TestJDBCBackend {
List<StatisticEntity> statisticEntities = Lists.newArrayList();
StatisticEntity statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, table.nameIdentifier(), Entity.EntityType.TABLE);
statisticEntities.clear();
statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, topic.nameIdentifier(), Entity.EntityType.TOPIC);
statisticEntities.clear();
statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, fileset.nameIdentifier(),
Entity.EntityType.FILESET);
statisticEntities.clear();
statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, model.nameIdentifier(), Entity.EntityType.MODEL);
// assert stats
@@ -459,25 +1209,25 @@ public class TestStatisticMetaService extends
TestJDBCBackend {
statisticEntities.clear();
statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, table.nameIdentifier(), Entity.EntityType.TABLE);
statisticEntities.clear();
statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, topic.nameIdentifier(), Entity.EntityType.TOPIC);
statisticEntities.clear();
statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, fileset.nameIdentifier(),
Entity.EntityType.FILESET);
statisticEntities.clear();
statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, model.nameIdentifier(), Entity.EntityType.MODEL);
// assert stats
@@ -540,25 +1290,25 @@ public class TestStatisticMetaService extends
TestJDBCBackend {
statisticEntities = Lists.newArrayList();
statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, table.nameIdentifier(), Entity.EntityType.TABLE);
statisticEntities.clear();
statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, topic.nameIdentifier(), Entity.EntityType.TOPIC);
statisticEntities.clear();
statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, fileset.nameIdentifier(),
Entity.EntityType.FILESET);
statisticEntities.clear();
statisticEntity = createStatisticEntity(auditInfo, 100L);
statisticEntities.add(statisticEntity);
- statisticMetaService.batchInsertStatisticPOsOnDuplicateKeyUpdate(
+ statisticMetaService.writeStatisticsWithVersion(
statisticEntities, model.nameIdentifier(), Entity.EntityType.MODEL);
// assert stats count
diff --git a/docs/manage-statistics-in-gravitino.md
b/docs/manage-statistics-in-gravitino.md
index 05676e8fba..754915ae77 100644
--- a/docs/manage-statistics-in-gravitino.md
+++ b/docs/manage-statistics-in-gravitino.md
@@ -24,6 +24,19 @@ may reserve later.
Updating creates a statistic that does not exist and overwrites one that does.
Reserved statistics
maintained by the system are not modifiable and the request is rejected.
+Concurrent creation, modification or deletion of the same statistic can return
HTTP 409, as can a
+write that times out waiting for a concurrent writer. The entire update batch
is rolled back on a
+conflict. If the client chooses to retry, it should retry against the current
state. If the target
+table was deleted or replaced while the operation was in progress, the request
fails with HTTP 404.
+On a MySQL backend, whose collation ignores trailing spaces, a name that
differs only in trailing
+spaces from an existing statistic, or from another name in the same request,
is rejected with
+HTTP 400.
+
+OCC conflicts are an expected result of concurrent writes. Clients decide
whether to abort or
+retry according to their application semantics. Gravitino does not
automatically retry a
+conflicting write. The built-in statistics updater propagates the exception to
its caller; a
+maintenance job can therefore fail on a conflict under its existing failure
handling.
+
<Tabs groupId='language' queryString>
<TabItem value="shell" label="REST">
@@ -74,6 +87,11 @@ List<Statistic> statistics =
orders.supportsStatistics().listStatistics();
### Drop Statistics
+Dropping a statistic that another request has already removed is not a
conflict. If a statistic
+was updated or deleted and recreated after the drop read it, the request can
return HTTP 409 and
+the entire drop batch is rolled back. Retry against the current state. If the
target table was
+deleted or replaced while dropping existing statistics, the request fails with
HTTP 404.
+
<Tabs groupId='language' queryString>
<TabItem value="shell" label="REST">
diff --git a/docs/open-api/statistics.yaml b/docs/open-api/statistics.yaml
index 826a4b7ec1..e373d8c36c 100644
--- a/docs/open-api/statistics.yaml
+++ b/docs/open-api/statistics.yaml
@@ -97,7 +97,7 @@ paths:
NoSuchMetadataObjectException:
$ref: "#/components/examples/NoSuchMetadataObjectException"
"409":
- description: Conflict - The specified statistic is unmodifiable
+ description: Conflict - The specified statistic is unmodifiable or
was modified concurrently; the caller decides whether to abort or retry against
the current state
content:
application/vnd.gravitino.v1+json:
schema:
@@ -150,7 +150,7 @@ paths:
NoSuchMetadataObjectException:
$ref: "#/components/examples/NoSuchMetadataObjectException"
"409":
- description: Conflict - The specified statistic is unmodifiable
+ description: Conflict - The specified statistic is unmodifiable or
was modified concurrently; the caller decides whether to abort or retry against
the current state
content:
application/vnd.gravitino.v1+json:
schema:
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/updater/statistics/TestGravitinoStatisticsUpdater.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/updater/statistics/TestGravitinoStatisticsUpdater.java
index f856cc7c03..7780d021f9 100644
---
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/updater/statistics/TestGravitinoStatisticsUpdater.java
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/updater/statistics/TestGravitinoStatisticsUpdater.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Map;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.maintenance.optimizer.api.common.PartitionPath;
import org.apache.gravitino.maintenance.optimizer.api.common.StatisticEntry;
import org.apache.gravitino.maintenance.optimizer.common.PartitionEntryImpl;
@@ -32,6 +33,7 @@ import
org.apache.gravitino.maintenance.optimizer.recommender.util.PartitionUtil
import org.apache.gravitino.stats.PartitionStatisticsUpdate;
import org.apache.gravitino.stats.StatisticValue;
import org.apache.gravitino.stats.StatisticValues;
+import org.apache.gravitino.stats.SupportsStatistics;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
@@ -124,6 +126,30 @@ class TestGravitinoStatisticsUpdater {
Assertions.assertEquals(3L, updates.get(0).statistics().get("s2").value());
}
+ @Test
+ void testTableStatisticConflictPropagatesWithoutRetry() {
+ GravitinoStatisticsUpdater updater = new GravitinoStatisticsUpdater();
+ GravitinoClient client = Mockito.mock(GravitinoClient.class,
Mockito.RETURNS_DEEP_STUBS);
+ updater.setGravitinoClientForTest(client);
+ SupportsStatistics statistics =
+ client
+ .loadCatalog("catalog")
+ .asTableCatalog()
+ .loadTable(NameIdentifier.of("db", "table"))
+ .supportsStatistics();
+ OptimisticLockException conflict = new OptimisticLockException("concurrent
statistic update");
+
Mockito.doThrow(conflict).when(statistics).updateStatistics(Mockito.anyMap());
+
+ Assertions.assertSame(
+ conflict,
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ updater.updateTableStatistics(
+ NameIdentifier.of("catalog", "db", "table"),
List.of(stat("row_count", 10L)))));
+ Mockito.verify(statistics).updateStatistics(Mockito.anyMap());
+ }
+
private StatisticEntry<?> stat(String name, long value) {
return new StatisticEntryImpl<>(name, StatisticValues.longValue(value));
}
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestSemanticModelOperations.java
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestSemanticModelOperations.java
index 07d7c7ca50..c5d7f6b098 100644
---
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestSemanticModelOperations.java
+++
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestSemanticModelOperations.java
@@ -1056,7 +1056,6 @@ public class TestSemanticModelOperations extends
BaseOperationsTest {
JsonUtils.objectMapper().readValue(response.body(),
SemanticModelResponse.class);
body.validate();
Assertions.assertEquals("marketing", body.getSemanticModel().name());
- verify(dispatcher).importOssieDocument(namespace,
OssieDocument.yaml(yaml));
verify(dispatcher)
.createSemanticModel(
eq(ident),
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestStatisticOperations.java
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestStatisticOperations.java
index 751ee690c2..f8654506c5 100644
---
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestStatisticOperations.java
+++
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestStatisticOperations.java
@@ -61,6 +61,7 @@ import org.apache.gravitino.dto.stats.StatisticDTO;
import org.apache.gravitino.dto.util.DTOConverters;
import org.apache.gravitino.exceptions.IllegalStatisticNameException;
import org.apache.gravitino.exceptions.NoSuchMetadataObjectException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.exceptions.UnmodifiableStatisticException;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.meta.AuditInfo;
@@ -516,6 +517,52 @@ public class TestStatisticOperations extends
BaseOperationsTest {
Assertions.assertFalse(errorResponse.getMessage().contains(longName));
}
+ @Test
+ public void testUpdateAndDropTableStatisticsReportOptimisticLockConflict() {
+ MetadataObject tableObject =
+ MetadataObjects.parse(
+ String.format("%s.%s.%s", catalog, schema, table),
MetadataObject.Type.TABLE);
+ String path =
+ "/metalakes/"
+ + metalake
+ + "/objects/"
+ + tableObject.type()
+ + "/"
+ + tableObject.fullName()
+ + "/statistics";
+ when(tableDispatcher.tableExists(any())).thenReturn(true);
+ OptimisticLockException conflict =
+ new OptimisticLockException("statistic was modified concurrently");
+
+ Map<String, StatisticValue<?>> statsMap = Maps.newHashMap();
+ statsMap.put(Statistic.CUSTOM_PREFIX + "test1",
StatisticValues.longValue(1L));
+ doThrow(conflict).when(manager).updateStatistics(any(), any(), any());
+ Response updateResp =
+ target(path)
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .put(entity(new StatisticsUpdateRequest(statsMap),
MediaType.APPLICATION_JSON_TYPE));
+ assertOptimisticLockConflict(updateResp);
+
+ doThrow(conflict).when(manager).dropStatistics(any(), any(), any());
+ Response dropResp =
+ target(path)
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .post(
+ entity(
+ new StatisticsDropRequest(new String[]
{Statistic.CUSTOM_PREFIX + "test1"}),
+ MediaType.APPLICATION_JSON_TYPE));
+ assertOptimisticLockConflict(dropResp);
+ }
+
+ private static void assertOptimisticLockConflict(Response resp) {
+ Assertions.assertEquals(Response.Status.CONFLICT.getStatusCode(),
resp.getStatus());
+ ErrorResponse errorResp = resp.readEntity(ErrorResponse.class);
+ Assertions.assertEquals(ErrorConstants.OPTIMISTIC_LOCK_CONFLICT_CODE,
errorResp.getCode());
+ Assertions.assertEquals(OptimisticLockException.class.getSimpleName(),
errorResp.getType());
+ }
+
@Test
public void testUpdateTableStatisticsWithNullRequestBody() {
when(tableDispatcher.tableExists(any())).thenReturn(true);
