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 5388510e62 [#13894] fix(core): Read entity aggregates from consistent
statement snapshots (#13880)
5388510e62 is described below
commit 5388510e628b5530b521f2522656619bfc5e760e
Author: Qi Yu <[email protected]>
AuthorDate: Tue Oct 27 19:50:50 2026 +0800
[#13894] fix(core): Read entity aggregates from consistent statement
snapshots (#13880)
### What changes were proposed in this pull request?
Load table columns, role privileges, and user/group memberships together
with their parent metadata in one SELECT, including updater inputs.
Distinguish unloaded PO collections from loaded empty snapshots, use
checked aggregate accessors, and keep nested result maps in XML rather
than placeholder mapper methods. Preserve loaded column snapshots in
TablePO copies.
### Why are the changes needed?
Concurrent overwrite/drop between separate queries can produce
incomplete or mixed-version entities and populate the cache with them.
Fix: #13894
Part of #13853: the read-consistency P0 only. The other epic tasks
remain open.
### Does this PR introduce _any_ user-facing change?
Gets and updater inputs receive complete aggregate snapshots. No REST,
configuration or schema changes. Role target-name resolution and
cross-node cache invalidation retain their existing semantics.
The internal `listSecurableObjectsByRoleId` service metric is no longer
emitted; role reads use the enclosing aggregate-read operation. The
full-name table query deliberately repeats the parent/root joins to keep
history aggregation bounded and portable. This is a documented cost
tradeoff, without a full-name throughput claim.
### How was this patch tested?
- Review follow-up: 165 targeted tests passed: 110 H2 backend cases and
55 unit tests, with Spotless checks. No unfiltered core test suite was
run.
- The 21 snapshot/guard cases include multi-column reads, concurrent
overwrite/drop, updater inputs, cache population, duplicate privilege
targets, and metadata-only versus loaded-empty snapshots. Copy tests use
a non-empty column snapshot and verify independent lists.
- Mutation checks detect split reads, column child-identity collapse,
missing metadata-only guards, a missing column-snapshot copy, and a
missing static-root SQL guard. The final fixed source passes again after
restoration.
- The original SQL was previously exercised on H2, MySQL 8 and
PostgreSQL 13. This follow-up reran H2 only because Docker is
unavailable; MySQL/PostgreSQL need CI revalidation for the updated
mappings.
- The earlier 45,000-history-row EXPLAIN checks establish targeted
history aggregation/index lookup, not a throughput benchmark or a
measured cost for the full-name variant.
---
.../storage/relational/mapper/GroupMetaMapper.java | 33 +
.../mapper/GroupMetaSQLProviderFactory.java | 11 +
.../storage/relational/mapper/RoleMetaMapper.java | 49 ++
.../mapper/RoleMetaSQLProviderFactory.java | 25 +
.../storage/relational/mapper/TableMetaMapper.java | 63 ++
.../mapper/TableMetaSQLProviderFactory.java | 30 +
.../storage/relational/mapper/UserMetaMapper.java | 33 +
.../mapper/UserMetaSQLProviderFactory.java | 11 +
.../provider/base/GroupMetaBaseSQLProvider.java | 21 +
.../provider/base/RoleMetaBaseSQLProvider.java | 43 +-
.../provider/base/TableMetaBaseSQLProvider.java | 82 +++
.../provider/base/UserMetaBaseSQLProvider.java | 21 +
.../gravitino/storage/relational/po/GroupPO.java | 28 +
.../gravitino/storage/relational/po/RolePO.java | 28 +
.../storage/relational/po/SecurableObjectPO.java | 16 +
.../gravitino/storage/relational/po/TablePO.java | 34 +-
.../gravitino/storage/relational/po/UserPO.java | 28 +
.../relational/service/GroupMetaService.java | 19 +-
.../relational/service/RoleMetaService.java | 32 +-
.../relational/service/TableMetaService.java | 26 +-
.../relational/service/TablePOStorageOps.java | 27 +-
.../relational/service/UserMetaService.java | 17 +-
.../storage/relational/mapper/RoleMetaMapper.xml | 43 ++
.../storage/relational/mapper/TableMetaMapper.xml | 40 ++
.../base/TestTableMetaBaseSQLProvider.java | 31 +
.../storage/relational/po/TestTablePO.java | 34 +
.../service/TestConsistentEntityReads.java | 774 +++++++++++++++++++++
.../relational/service/TestRoleMetaService.java | 7 +-
design-docs/entity-store-consistent-reads-audit.md | 71 ++
29 files changed, 1632 insertions(+), 45 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java
index 333be53597..fec8c21e23 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java
@@ -26,7 +26,10 @@ import org.apache.gravitino.storage.relational.po.GroupPO;
import org.apache.gravitino.storage.relational.po.auth.GroupUpdatedAt;
import org.apache.ibatis.annotations.DeleteProvider;
import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Many;
import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Result;
+import org.apache.ibatis.annotations.Results;
import org.apache.ibatis.annotations.SelectProvider;
import org.apache.ibatis.annotations.UpdateProvider;
@@ -137,4 +140,34 @@ public interface GroupMetaMapper {
@SelectProvider(type = GroupMetaSQLProviderFactory.class, method =
"getGroupUpdatedAt")
GroupUpdatedAt getGroupUpdatedAt(
@Param("metalakeName") String metalakeName, @Param("groupName") String
groupName);
+ /**
+ * Reads a principal and its live role memberships from one statement
snapshot.
+ *
+ * @param metalakeId the parent metalake ID
+ * @param groupName the principal name
+ * @return the principal snapshot, or null when it does not exist
+ */
+ @Nullable
+ @Results({
+ @Result(property = "groupId", column = "groupId", id = true),
+ @Result(property = "groupName", column = "groupName"),
+ @Result(property = "metalakeId", column = "metalakeId"),
+ @Result(property = "auditInfo", column = "auditInfo"),
+ @Result(property = "currentVersion", column = "currentVersion"),
+ @Result(property = "lastVersion", column = "lastVersion"),
+ @Result(property = "deletedAt", column = "deletedAt"),
+ @Result(
+ property = "rolePOs",
+ javaType = List.class,
+ many =
+ @Many(
+ resultMap =
+
"org.apache.gravitino.storage.relational.mapper.RoleMetaMapper.membershipRoleSnapshot",
+ columnPrefix = "role_"))
+ })
+ @SelectProvider(
+ type = GroupMetaSQLProviderFactory.class,
+ method = "selectGroupWithRolesByMetalakeIdAndName")
+ GroupPO selectGroupWithRolesByMetalakeIdAndName(
+ @Param("metalakeId") Long metalakeId, @Param("groupName") String
groupName);
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java
index 215b4083c6..0b812bc383 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java
@@ -132,4 +132,15 @@ public class GroupMetaSQLProviderFactory {
@Param("metalakeName") String metalakeName, @Param("groupName") String
groupName) {
return getProvider().getGroupUpdatedAt(metalakeName, groupName);
}
+ /**
+ * Returns SQL for a principal and membership snapshot.
+ *
+ * @param metalakeId the parent metalake ID
+ * @param groupName the principal name
+ * @return the aggregate select SQL
+ */
+ public static String selectGroupWithRolesByMetalakeIdAndName(
+ @Param("metalakeId") Long metalakeId, @Param("groupName") String
groupName) {
+ return getProvider().selectGroupWithRolesByMetalakeIdAndName(metalakeId,
groupName);
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaMapper.java
index aca5c0d8d2..23ef98082b 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaMapper.java
@@ -25,7 +25,11 @@ import org.apache.gravitino.storage.relational.po.RolePO;
import org.apache.gravitino.storage.relational.po.auth.RoleUpdatedAt;
import org.apache.ibatis.annotations.DeleteProvider;
import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Many;
import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Result;
+import org.apache.ibatis.annotations.ResultMap;
+import org.apache.ibatis.annotations.Results;
import org.apache.ibatis.annotations.SelectProvider;
import org.apache.ibatis.annotations.UpdateProvider;
@@ -122,4 +126,49 @@ public interface RoleMetaMapper {
@SelectProvider(type = RoleMetaSQLProviderFactory.class, method =
"batchGetRoleUpdatedAt")
List<RoleUpdatedAt> batchGetRoleUpdatedAt(@Param("roleIds") List<Long>
roleIds);
+
+ /**
+ * Reads a role and its live privileges from one statement snapshot.
+ *
+ * @param metalakeId the parent metalake ID
+ * @param roleName the role name
+ * @return the role snapshot, or null when it does not exist
+ */
+ @Results(
+ id = "roleWithPrivileges",
+ value = {
+ @Result(property = "roleId", column = "roleId", id = true),
+ @Result(property = "roleName", column = "roleName"),
+ @Result(property = "metalakeId", column = "metalakeId"),
+ @Result(property = "properties", column = "properties"),
+ @Result(property = "auditInfo", column = "auditInfo"),
+ @Result(property = "currentVersion", column = "currentVersion"),
+ @Result(property = "lastVersion", column = "lastVersion"),
+ @Result(property = "deletedAt", column = "deletedAt"),
+ @Result(
+ property = "securableObjectPOs",
+ javaType = List.class,
+ many = @Many(resultMap = "rolePrivilegeSnapshot", columnPrefix =
"priv_"))
+ })
+ @SelectProvider(
+ type = RoleMetaSQLProviderFactory.class,
+ method = "selectRoleWithPrivilegesByMetalakeIdAndName")
+ @Nullable
+ RolePO selectRoleWithPrivilegesByMetalakeIdAndName(
+ @Param("metalakeId") Long metalakeId, @Param("roleName") String
roleName);
+
+ /**
+ * Lists matching roles and all their privileges from one statement snapshot.
+ *
+ * @param metadataObjectId the metadata object ID
+ * @param metadataObjectType the metadata object type
+ * @return the role snapshots
+ */
+ @ResultMap("roleWithPrivileges")
+ @SelectProvider(
+ type = RoleMetaSQLProviderFactory.class,
+ method = "listRolesWithPrivilegesByMetadataObjectIdAndType")
+ List<RolePO> listRolesWithPrivilegesByMetadataObjectIdAndType(
+ @Param("metadataObjectId") Long metadataObjectId,
+ @Param("metadataObjectType") String metadataObjectType);
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaSQLProviderFactory.java
index 4d990b09ea..2ecb33ee3c 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaSQLProviderFactory.java
@@ -127,4 +127,29 @@ public class RoleMetaSQLProviderFactory {
public static String batchGetRoleUpdatedAt(@Param("roleIds") List<Long>
roleIds) {
return getProvider().batchGetRoleUpdatedAt(roleIds);
}
+ /**
+ * Returns SQL for a role and privilege snapshot by name.
+ *
+ * @param metalakeId the metalake ID
+ * @param roleName the role name
+ * @return the aggregate select SQL
+ */
+ public static String selectRoleWithPrivilegesByMetalakeIdAndName(
+ @Param("metalakeId") Long metalakeId, @Param("roleName") String
roleName) {
+ return
getProvider().selectRoleWithPrivilegesByMetalakeIdAndName(metalakeId, roleName);
+ }
+
+ /**
+ * Returns SQL for matching roles and all their privileges.
+ *
+ * @param metadataObjectId the metadata object ID
+ * @param metadataObjectType the metadata object type
+ * @return the aggregate select SQL
+ */
+ public static String listRolesWithPrivilegesByMetadataObjectIdAndType(
+ @Param("metadataObjectId") Long metadataObjectId,
+ @Param("metadataObjectType") String metadataObjectType) {
+ return getProvider()
+ .listRolesWithPrivilegesByMetadataObjectIdAndType(metadataObjectId,
metadataObjectType);
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.java
index 4fdc5103ae..be36fe45d7 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.java
@@ -20,10 +20,15 @@
package org.apache.gravitino.storage.relational.mapper;
import java.util.List;
+import javax.annotation.Nullable;
import org.apache.gravitino.storage.relational.po.TablePO;
import org.apache.ibatis.annotations.DeleteProvider;
import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Many;
import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Result;
+import org.apache.ibatis.annotations.ResultMap;
+import org.apache.ibatis.annotations.Results;
import org.apache.ibatis.annotations.SelectProvider;
import org.apache.ibatis.annotations.UpdateProvider;
@@ -38,6 +43,64 @@ import org.apache.ibatis.annotations.UpdateProvider;
public interface TableMetaMapper {
String TABLE_NAME = "table_meta";
+ /**
+ * Reads table metadata, its current version, and columns from one database
statement snapshot.
+ *
+ * @param schemaId the parent schema ID
+ * @param tableName the table name
+ * @return the aggregate snapshot, or null when the table does not exist
+ */
+ @Results(
+ id = "tableWithColumns",
+ value = {
+ @Result(property = "tableId", column = "tableId", id = true),
+ @Result(property = "tableName", column = "tableName"),
+ @Result(property = "metalakeId", column = "metalakeId"),
+ @Result(property = "catalogId", column = "catalogId"),
+ @Result(property = "schemaId", column = "schemaId"),
+ @Result(property = "auditInfo", column = "auditInfo"),
+ @Result(property = "currentVersion", column = "currentVersion"),
+ @Result(property = "lastVersion", column = "lastVersion"),
+ @Result(property = "deletedAt", column = "deletedAt"),
+ @Result(property = "format", column = "format"),
+ @Result(property = "properties", column = "properties"),
+ @Result(property = "partitions", column = "partitions"),
+ @Result(property = "sortOrders", column = "sortOrders"),
+ @Result(property = "distribution", column = "distribution"),
+ @Result(property = "indexes", column = "indexes"),
+ @Result(property = "comment", column = "comment"),
+ @Result(
+ property = "columnPOs",
+ javaType = List.class,
+ many = @Many(resultMap = "tableColumnSnapshot", columnPrefix =
"col_"))
+ })
+ @SelectProvider(
+ type = TableMetaSQLProviderFactory.class,
+ method = "selectTableWithColumnsBySchemaIdAndName")
+ @Nullable
+ TablePO selectTableWithColumnsBySchemaIdAndName(
+ @Param("schemaId") Long schemaId, @Param("tableName") String tableName);
+
+ /**
+ * Reads a table aggregate by its full name, retaining missing-parent
placeholders.
+ *
+ * @param metalakeName the metalake name
+ * @param catalogName the catalog name
+ * @param schemaName the physical schema name
+ * @param tableName the table name
+ * @return the aggregate or parent placeholder, or null when the metalake or
catalog is absent
+ */
+ @ResultMap("tableWithColumns")
+ @SelectProvider(
+ type = TableMetaSQLProviderFactory.class,
+ method = "selectTableWithColumnsByFullQualifiedName")
+ @Nullable
+ TablePO selectTableWithColumnsByFullQualifiedName(
+ @Param("metalakeName") String metalakeName,
+ @Param("catalogName") String catalogName,
+ @Param("schemaName") String schemaName,
+ @Param("tableName") String tableName);
+
@SelectProvider(type = TableMetaSQLProviderFactory.class, method =
"listTablePOsBySchemaId")
List<TablePO> listTablePOsBySchemaId(@Param("schemaId") Long schemaId);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaSQLProviderFactory.java
index 2876bbc1d1..10ea2fd2a0 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaSQLProviderFactory.java
@@ -147,4 +147,34 @@ public class TableMetaSQLProviderFactory {
@Param("schemaId") Long schemaId, @Param("tableNames") List<String>
tableNames) {
return getProvider().batchSelectTableByIdentifier(schemaId, tableNames);
}
+ /**
+ * Returns the single-statement table snapshot SQL for a schema ID and name.
+ *
+ * @param schemaId the parent schema ID
+ * @param tableName the table name
+ * @return the aggregate select SQL
+ */
+ public static String selectTableWithColumnsBySchemaIdAndName(
+ @Param("schemaId") Long schemaId, @Param("tableName") String tableName) {
+ return getProvider().selectTableWithColumnsBySchemaIdAndName(schemaId,
tableName);
+ }
+
+ /**
+ * Returns the single-statement table snapshot SQL for a full name.
+ *
+ * @param metalakeName the metalake name
+ * @param catalogName the catalog name
+ * @param schemaName the physical schema name
+ * @param tableName the table name
+ * @return the aggregate select SQL
+ */
+ public static String selectTableWithColumnsByFullQualifiedName(
+ @Param("metalakeName") String metalakeName,
+ @Param("catalogName") String catalogName,
+ @Param("schemaName") String schemaName,
+ @Param("tableName") String tableName) {
+ return getProvider()
+ .selectTableWithColumnsByFullQualifiedName(
+ metalakeName, catalogName, schemaName, tableName);
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java
index 4a10ac8dca..3e90371d00 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java
@@ -27,7 +27,10 @@ import
org.apache.gravitino.storage.relational.po.auth.AuthPrefetchRow;
import org.apache.gravitino.storage.relational.po.auth.UserUpdatedAt;
import org.apache.ibatis.annotations.DeleteProvider;
import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Many;
import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Result;
+import org.apache.ibatis.annotations.Results;
import org.apache.ibatis.annotations.SelectProvider;
import org.apache.ibatis.annotations.UpdateProvider;
@@ -163,4 +166,34 @@ public interface UserMetaMapper {
@Param("metalakeName") String metalakeName,
@Param("userName") String userName,
@Param("groupNames") List<String> groupNames);
+ /**
+ * Reads a principal and its live role memberships from one statement
snapshot.
+ *
+ * @param metalakeId the parent metalake ID
+ * @param userName the principal name
+ * @return the principal snapshot, or null when it does not exist
+ */
+ @Nullable
+ @Results({
+ @Result(property = "userId", column = "userId", id = true),
+ @Result(property = "userName", column = "userName"),
+ @Result(property = "metalakeId", column = "metalakeId"),
+ @Result(property = "auditInfo", column = "auditInfo"),
+ @Result(property = "currentVersion", column = "currentVersion"),
+ @Result(property = "lastVersion", column = "lastVersion"),
+ @Result(property = "deletedAt", column = "deletedAt"),
+ @Result(
+ property = "rolePOs",
+ javaType = List.class,
+ many =
+ @Many(
+ resultMap =
+
"org.apache.gravitino.storage.relational.mapper.RoleMetaMapper.membershipRoleSnapshot",
+ columnPrefix = "role_"))
+ })
+ @SelectProvider(
+ type = UserMetaSQLProviderFactory.class,
+ method = "selectUserWithRolesByMetalakeIdAndName")
+ UserPO selectUserWithRolesByMetalakeIdAndName(
+ @Param("metalakeId") Long metalakeId, @Param("userName") String
userName);
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java
index bc52e64d52..1962288ee9 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java
@@ -136,4 +136,15 @@ public class UserMetaSQLProviderFactory {
@Param("groupNames") List<String> groupNames) {
return getProvider().batchGetAuthSubjectsForUser(metalakeName, userName,
groupNames);
}
+ /**
+ * Returns SQL for a principal and membership snapshot.
+ *
+ * @param metalakeId the parent metalake ID
+ * @param userName the principal name
+ * @return the aggregate select SQL
+ */
+ public static String selectUserWithRolesByMetalakeIdAndName(
+ @Param("metalakeId") Long metalakeId, @Param("userName") String
userName) {
+ return getProvider().selectUserWithRolesByMetalakeIdAndName(metalakeId,
userName);
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java
index 4afa7ad3d2..cec0822680 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java
@@ -316,4 +316,25 @@ public class GroupMetaBaseSQLProvider {
+ " WHERE mm.metalake_name = #{metalakeName} AND gm.group_name =
#{groupName}"
+ " AND gm.deleted_at = 0";
}
+ /**
+ * Returns a single-statement principal and membership snapshot.
+ *
+ * @param metalakeId the metalake ID
+ * @param groupName the principal name
+ * @return the aggregate select SQL
+ */
+ public String selectGroupWithRolesByMetalakeIdAndName(
+ @Param("metalakeId") Long metalakeId, @Param("groupName") String
groupName) {
+ return "SELECT p.*, r.role_id AS role_roleId, r.role_name AS
role_roleName,"
+ + " r.metalake_id AS role_metalakeId, r.properties AS role_properties,"
+ + " r.audit_info AS role_auditInfo, r.current_version AS
role_currentVersion,"
+ + " r.last_version AS role_lastVersion, r.deleted_at AS role_deletedAt
FROM ("
+ + selectGroupMetaByMetalakeIdAndName(metalakeId, groupName)
+ + ") p LEFT JOIN "
+ + GROUP_ROLE_RELATION_TABLE_NAME
+ + " rel"
+ + " ON rel.group_id = p.groupId AND rel.deleted_at = 0 LEFT JOIN "
+ + ROLE_TABLE_NAME
+ + " r ON r.role_id = rel.role_id AND r.deleted_at = 0";
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/RoleMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/RoleMetaBaseSQLProvider.java
index 17083abc43..d44e4ef71c 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/RoleMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/RoleMetaBaseSQLProvider.java
@@ -232,7 +232,46 @@ public class RoleMetaBaseSQLProvider {
}
return "<script>SELECT role_id as roleId, role_name as roleName,
updated_at as updatedAt FROM "
+ ROLE_TABLE_NAME
- + " WHERE role_id IN <foreach item='id' collection='roleIds' open='('
separator=',' close=')'>#{id}</foreach>"
- + " AND deleted_at = 0</script>";
+ + " WHERE role_id IN <foreach item='id' collection='roleIds' open='('
separator=','"
+ + " close=')'>#{id}</foreach> AND deleted_at = 0</script>";
+ }
+ /**
+ * Returns a single-statement role and privilege snapshot by name.
+ *
+ * @param metalakeId the metalake ID
+ * @param roleName the role name
+ * @return the aggregate select SQL
+ */
+ public String selectRoleWithPrivilegesByMetalakeIdAndName(
+ @Param("metalakeId") Long metalakeId, @Param("roleName") String
roleName) {
+ return withPrivileges(selectRoleMetaByMetalakeIdAndName(metalakeId,
roleName));
+ }
+
+ /**
+ * Returns matching roles together with all their privileges.
+ *
+ * @param metadataObjectId the metadata object ID
+ * @param metadataObjectType the metadata object type
+ * @return the aggregate select SQL
+ */
+ public String listRolesWithPrivilegesByMetadataObjectIdAndType(
+ @Param("metadataObjectId") Long metadataObjectId,
+ @Param("metadataObjectType") String metadataObjectType) {
+ return withPrivileges(listRolesByMetadataObjectIdAndType(metadataObjectId,
metadataObjectType));
+ }
+
+ private String withPrivileges(String roleSql) {
+ // Role overwrites replace their privilege rows. Resolve both from one
statement snapshot,
+ // including empty roles, rather than combining a root read with a later
privilege read.
+ return "SELECT r.*, p.id AS priv_relationId, p.role_id AS priv_roleId,
p.metadata_object_id AS"
+ + " priv_metadataObjectId, p.type AS priv_type, p.privilege_names AS"
+ + " priv_privilegeNames, p.privilege_conditions AS
priv_privilegeConditions,"
+ + " p.current_version AS priv_currentVersion, p.last_version AS
priv_lastVersion,"
+ + " p.deleted_at AS priv_deletedAt FROM ("
+ + roleSql
+ + ") r LEFT JOIN "
+ + SecurableObjectMapper.SECURABLE_OBJECT_TABLE_NAME
+ + " p"
+ + " ON r.roleId = p.role_id AND p.deleted_at = 0";
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TableMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TableMetaBaseSQLProvider.java
index 32b45b590d..0b31b3f239 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TableMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TableMetaBaseSQLProvider.java
@@ -20,12 +20,15 @@ package
org.apache.gravitino.storage.relational.mapper.provider.base;
import static
org.apache.gravitino.storage.relational.mapper.TableMetaMapper.TABLE_NAME;
+import com.google.common.base.Preconditions;
import java.util.List;
import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
import org.apache.gravitino.storage.relational.mapper.TableVersionMapper;
import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
+import org.apache.gravitino.storage.relational.po.ColumnPO;
import org.apache.gravitino.storage.relational.po.TablePO;
import org.apache.ibatis.annotations.Param;
@@ -426,4 +429,83 @@ public class TableMetaBaseSQLProvider {
"""
.formatted(TABLE_NAME, TableVersionMapper.TABLE_NAME);
}
+ /**
+ * Builds a single-statement table and column snapshot by schema ID and name.
+ *
+ * @param schemaId the schema ID
+ * @param tableName the table name
+ * @return the aggregate select SQL
+ */
+ public String selectTableWithColumnsBySchemaIdAndName(
+ @Param("schemaId") Long schemaId, @Param("tableName") String tableName) {
+ return withColumns(selectTableMetaBySchemaIdAndName(schemaId, tableName));
+ }
+
+ /**
+ * Builds a single-statement table and column snapshot by full name.
+ *
+ * @param metalakeName the metalake name
+ * @param catalogName the catalog name
+ * @param schemaName the physical schema name
+ * @param tableName the table name
+ * @return the aggregate select SQL
+ */
+ public String selectTableWithColumnsByFullQualifiedName(
+ @Param("metalakeName") String metalakeName,
+ @Param("catalogName") String catalogName,
+ @Param("schemaName") String schemaName,
+ @Param("tableName") String tableName) {
+ return withColumns(
+ selectTableByFullQualifiedName(metalakeName, catalogName, schemaName,
tableName));
+ }
+
+ private String withColumns(String tableSql) {
+ // One statement gives the root, version and columns a common snapshot.
Overwrite retires old
+ // columns, so a second SELECT cannot reconstruct them. Aggregate history
once and join by the
+ // existing (table_id, table_version, column_id, deleted_at) unique key.
Correlated scans for
+ // each column would be quadratic for long histories. Root subqueries have
independent bound
+ // parameters on all supported backends. Exclude DELETE markers only after
resolving history,
+ // otherwise a dropped column would reappear through its earlier
CREATE/UPDATE row.
+ // Derived tables accept static SELECTs only. MyBatis handles <script>
only at the start of
+ // the complete statement, so a dynamic provider cannot be embedded here.
+ Preconditions.checkArgument(
+ !tableSql.stripLeading().startsWith("<script>"),
+ "Aggregate table reads require static root SQL, not a MyBatis
<script>");
+ String rootSql = tableSql.strip().replaceFirst(";\\s*$", "");
+ return "SELECT t.*, "
+ + "c.column_id AS col_columnId, "
+ + "c.column_name AS col_columnName, "
+ + "c.column_position AS col_columnPosition, "
+ + "c.metalake_id AS col_metalakeId, "
+ + "c.catalog_id AS col_catalogId, "
+ + "c.schema_id AS col_schemaId, "
+ + "c.table_id AS col_tableId, "
+ + "c.table_version AS col_tableVersion, "
+ + "c.column_type AS col_columnType, "
+ + "c.column_comment AS col_columnComment, "
+ + "c.column_nullable AS col_nullable, "
+ + "c.column_auto_increment AS col_autoIncrement, "
+ + "c.column_default_value AS col_defaultValue, "
+ + "c.column_op_type AS col_columnOpType, "
+ + "c.deleted_at AS col_deletedAt, "
+ + "c.audit_info AS col_auditInfo "
+ + "FROM ("
+ + rootSql
+ + ") t LEFT JOIN ("
+ + " SELECT history.table_id, history.column_id,
MAX(history.table_version) AS max_version"
+ + " FROM "
+ + TableColumnMapper.COLUMN_TABLE_NAME
+ + " history JOIN ("
+ + rootSql
+ + ") observed"
+ + " ON history.table_id = observed.tableId AND history.table_version
<= observed.currentVersion"
+ + " WHERE history.deleted_at = 0 GROUP BY history.table_id,
history.column_id"
+ + ") latest ON latest.table_id = t.tableId LEFT JOIN "
+ + TableColumnMapper.COLUMN_TABLE_NAME
+ + " c"
+ + " ON c.table_id = latest.table_id AND c.column_id = latest.column_id"
+ + " AND c.table_version = latest.max_version AND c.deleted_at = 0"
+ + " AND c.column_op_type != "
+ + ColumnPO.ColumnOpType.DELETE.value();
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java
index dbe362b7db..cb15a292cb 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java
@@ -392,4 +392,25 @@ public class UserMetaBaseSQLProvider {
return "<script>" + userBranch + userRoleBranch + groupBranch +
groupRoleBranch + "</script>";
}
+ /**
+ * Returns a single-statement principal and membership snapshot.
+ *
+ * @param metalakeId the metalake ID
+ * @param userName the principal name
+ * @return the aggregate select SQL
+ */
+ public String selectUserWithRolesByMetalakeIdAndName(
+ @Param("metalakeId") Long metalakeId, @Param("userName") String
userName) {
+ return "SELECT p.*, r.role_id AS role_roleId, r.role_name AS
role_roleName,"
+ + " r.metalake_id AS role_metalakeId, r.properties AS role_properties,"
+ + " r.audit_info AS role_auditInfo, r.current_version AS
role_currentVersion,"
+ + " r.last_version AS role_lastVersion, r.deleted_at AS role_deletedAt
FROM ("
+ + selectUserMetaByMetalakeIdAndName(metalakeId, userName)
+ + ") p LEFT JOIN "
+ + USER_ROLE_RELATION_TABLE_NAME
+ + " rel"
+ + " ON rel.user_id = p.userId AND rel.deleted_at = 0 LEFT JOIN "
+ + ROLE_TABLE_NAME
+ + " r ON r.role_id = rel.role_id AND r.deleted_at = 0";
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/GroupPO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/GroupPO.java
index 478e9e774c..fae78814b8 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/po/GroupPO.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/po/GroupPO.java
@@ -20,6 +20,8 @@ package org.apache.gravitino.storage.relational.po;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
+import java.util.List;
+import javax.annotation.Nullable;
public class GroupPO {
private Long groupId;
@@ -30,6 +32,32 @@ public class GroupPO {
private Long lastVersion;
private Long deletedAt;
+ @Nullable private List<RolePO> rolePOs;
+
+ /**
+ * Returns the membership rows loaded in the same statement as this entity.
+ *
+ * @return the snapshot, empty when loaded without children, or null for
metadata-only reads
+ */
+ @Nullable
+ public List<RolePO> getRolePOs() {
+ return rolePOs;
+ }
+
+ /**
+ * Requires the aggregate snapshot before a complete entity is assembled or
updated.
+ *
+ * @return the loaded membership snapshot, which may be empty
+ * @throws IllegalStateException if only the entity metadata was loaded
+ */
+ public List<RolePO> requireRolePOs() {
+ Preconditions.checkState(
+ rolePOs != null,
+ "The membership snapshot was not loaded for group %s; use an aggregate
mapper",
+ groupId);
+ return rolePOs;
+ }
+
public Long getGroupId() {
return groupId;
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/RolePO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/RolePO.java
index 9c59ec7151..34b5e7f37f 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/po/RolePO.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/po/RolePO.java
@@ -20,6 +20,8 @@ package org.apache.gravitino.storage.relational.po;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
+import java.util.List;
+import javax.annotation.Nullable;
public class RolePO {
private Long roleId;
@@ -31,6 +33,32 @@ public class RolePO {
private Long lastVersion;
private Long deletedAt;
+ @Nullable private List<SecurableObjectPO> securableObjectPOs;
+
+ /**
+ * Returns the privilege rows loaded in the same statement as this entity.
+ *
+ * @return the snapshot, empty when loaded without children, or null for
metadata-only reads
+ */
+ @Nullable
+ public List<SecurableObjectPO> getSecurableObjectPOs() {
+ return securableObjectPOs;
+ }
+
+ /**
+ * Requires the aggregate snapshot before a complete entity is assembled or
updated.
+ *
+ * @return the loaded privilege snapshot, which may be empty
+ * @throws IllegalStateException if only the entity metadata was loaded
+ */
+ public List<SecurableObjectPO> requireSecurableObjectPOs() {
+ Preconditions.checkState(
+ securableObjectPOs != null,
+ "The privilege snapshot was not loaded for role %s; use an aggregate
mapper",
+ roleId);
+ return securableObjectPOs;
+ }
+
public Long getRoleId() {
return roleId;
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/SecurableObjectPO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/SecurableObjectPO.java
index 42cf5b3868..ed1a45ff66 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/po/SecurableObjectPO.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/po/SecurableObjectPO.java
@@ -20,9 +20,12 @@ package org.apache.gravitino.storage.relational.po;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
+import javax.annotation.Nullable;
public class SecurableObjectPO {
+ @Nullable private Long id;
+
private Long roleId;
private Long metadataObjectId;
private String type;
@@ -32,6 +35,19 @@ public class SecurableObjectPO {
private Long lastVersion;
private Long deletedAt;
+ /**
+ * Returns the generated relation-row identity when it was selected by the
query.
+ *
+ * <p>The ID is not an insert input. A role can contain multiple stored rows
for the same target;
+ * aggregate result mappings must use this physical identity to preserve
every row.
+ *
+ * @return the stored row ID, or null when it was not selected
+ */
+ @Nullable
+ public Long getId() {
+ return id;
+ }
+
public Long getRoleId() {
return roleId;
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/TablePO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/TablePO.java
index 22d92a76b9..8a3c1fa960 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/po/TablePO.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/po/TablePO.java
@@ -20,6 +20,9 @@ package org.apache.gravitino.storage.relational.po;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.List;
+import javax.annotation.Nullable;
import lombok.Getter;
@Getter
@@ -43,6 +46,33 @@ public class TablePO {
private String indexes;
private String comment;
+ // Populated only by the single-statement aggregate read; not a column of
table_meta.
+ @Nullable private List<ColumnPO> columnPOs;
+
+ /**
+ * Returns the column rows loaded in the same statement as this entity.
+ *
+ * @return the snapshot, empty when loaded without children, or null for
metadata-only reads
+ */
+ @Nullable
+ public List<ColumnPO> getColumnPOs() {
+ return columnPOs;
+ }
+
+ /**
+ * Requires the aggregate snapshot before a complete entity is assembled or
updated.
+ *
+ * @return the loaded column snapshot, which may be empty
+ * @throws IllegalStateException if only the entity metadata was loaded
+ */
+ public List<ColumnPO> requireColumnPOs() {
+ Preconditions.checkState(
+ columnPOs != null,
+ "The column snapshot was not loaded for table %s; use an aggregate
mapper",
+ tableId);
+ return columnPOs;
+ }
+
public Long getTableId() {
return tableId;
}
@@ -140,6 +170,7 @@ public class TablePO {
tablePO.distribution = source.distribution;
tablePO.indexes = source.indexes;
tablePO.comment = source.comment;
+ tablePO.columnPOs = source.columnPOs == null ? null : new
ArrayList<>(source.columnPOs);
}
public Builder withTableId(Long tableId) {
@@ -251,7 +282,8 @@ public class TablePO {
/**
* Creates a new instance of {@link Builder} pre-filled with the values of
an existing table, so
- * that a caller changing a few columns does not have to restate the others.
+ * that a caller changing a few columns does not have to restate the others.
Loaded column
+ * snapshots are copied into an independent list; metadata-only copies
remain unloaded.
*
* @param tablePO The table to copy the values from.
* @return The new instance.
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/UserPO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/UserPO.java
index 2a9079bc53..eee257e58c 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/po/UserPO.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/po/UserPO.java
@@ -20,6 +20,8 @@ package org.apache.gravitino.storage.relational.po;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
+import java.util.List;
+import javax.annotation.Nullable;
public class UserPO {
private Long userId;
@@ -30,6 +32,32 @@ public class UserPO {
private Long lastVersion;
private Long deletedAt;
+ @Nullable private List<RolePO> rolePOs;
+
+ /**
+ * Returns the membership rows loaded in the same statement as this entity.
+ *
+ * @return the snapshot, empty when loaded without children, or null for
metadata-only reads
+ */
+ @Nullable
+ public List<RolePO> getRolePOs() {
+ return rolePOs;
+ }
+
+ /**
+ * Requires the aggregate snapshot before a complete entity is assembled or
updated.
+ *
+ * @return the loaded membership snapshot, which may be empty
+ * @throws IllegalStateException if only the entity metadata was loaded
+ */
+ public List<RolePO> requireRolePOs() {
+ Preconditions.checkState(
+ rolePOs != null,
+ "The membership snapshot was not loaded for user %s; use an aggregate
mapper",
+ userId);
+ return rolePOs;
+ }
+
public Long getUserId() {
return userId;
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/GroupMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/GroupMetaService.java
index 44087c7366..df06e6e4b9 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/GroupMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/GroupMetaService.java
@@ -66,10 +66,18 @@ public class GroupMetaService {
private GroupMetaService() {}
private GroupPO getGroupPOByMetalakeIdAndName(Long metalakeId, String
groupName) {
+ return getGroupPOByMetalakeIdAndName(metalakeId, groupName, false);
+ }
+
+ private GroupPO getGroupPOByMetalakeIdAndName(
+ Long metalakeId, String groupName, boolean withRoles) {
GroupPO GroupPO =
SessionUtils.getWithoutCommit(
GroupMetaMapper.class,
- mapper -> mapper.selectGroupMetaByMetalakeIdAndName(metalakeId,
groupName));
+ mapper ->
+ withRoles
+ ?
mapper.selectGroupWithRolesByMetalakeIdAndName(metalakeId, groupName)
+ : mapper.selectGroupMetaByMetalakeIdAndName(metalakeId,
groupName));
if (GroupPO == null) {
throw new NoSuchEntityException(
@@ -106,8 +114,8 @@ public class GroupMetaService {
NameIdentifier metalakeIdent =
NameIdentifier.of(NameIdentifierUtil.getMetalake(identifier));
long metalakeId = EntityIdService.getEntityId(metalakeIdent,
Entity.EntityType.METALAKE);
- GroupPO groupPO = getGroupPOByMetalakeIdAndName(metalakeId,
identifier.name());
- List<RolePO> rolePOs =
RoleMetaService.getInstance().listRolesByGroupId(groupPO.getGroupId());
+ GroupPO groupPO = getGroupPOByMetalakeIdAndName(metalakeId,
identifier.name(), true);
+ List<RolePO> rolePOs = groupPO.requireRolePOs();
return POConverters.fromGroupPO(groupPO, rolePOs, identifier.namespace());
}
@@ -280,9 +288,8 @@ public class GroupMetaService {
NameIdentifier metalakeIdent =
NameIdentifier.of(NameIdentifierUtil.getMetalake(identifier));
Long metalakeId = EntityIdService.getEntityId(metalakeIdent,
Entity.EntityType.METALAKE);
- GroupPO oldGroupPO = getGroupPOByMetalakeIdAndName(metalakeId,
identifier.name());
- List<RolePO> rolePOs =
-
RoleMetaService.getInstance().listRolesByGroupId(oldGroupPO.getGroupId());
+ GroupPO oldGroupPO = getGroupPOByMetalakeIdAndName(metalakeId,
identifier.name(), true);
+ List<RolePO> rolePOs = oldGroupPO.requireRolePOs();
GroupEntity oldGroupEntity =
POConverters.fromGroupPO(oldGroupPO, rolePOs, identifier.namespace());
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
index 040dcb1dfc..6a720db477 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
@@ -133,8 +133,11 @@ public class RoleMetaService {
SessionUtils.getWithoutCommit(
RoleMetaMapper.class,
mapper ->
- mapper.listRolesByMetadataObjectIdAndType(
- metadataObjectId, metadataObject.type().name()));
+ allFields
+ ? mapper.listRolesWithPrivilegesByMetadataObjectIdAndType(
+ metadataObjectId, metadataObject.type().name())
+ : mapper.listRolesByMetadataObjectIdAndType(
+ metadataObjectId, metadataObject.type().name()));
return rolePOs.stream()
.map(
po -> {
@@ -234,7 +237,7 @@ public class RoleMetaService {
String metalake = NameIdentifierUtil.getMetalake(identifier);
Long metalakeId =
MetalakeMetaService.getInstance().getMetalakeIdByName(metalake);
- RolePO rolePO = getRolePOByMetalakeIdAndName(metalakeId,
identifier.name());
+ RolePO rolePO = getRolePOByMetalakeIdAndName(metalakeId,
identifier.name(), true);
RoleEntity oldRoleEntity =
POConverters.fromRolePO(
rolePO, listSecurableObjects(rolePO),
AuthorizationUtils.ofRoleNamespace(metalake));
@@ -390,7 +393,7 @@ public class RoleMetaService {
Long metalakeId =
MetalakeMetaService.getInstance().getMetalakeIdByName(identifier.namespace().level(0));
- RolePO rolePO = getRolePOByMetalakeIdAndName(metalakeId,
identifier.name());
+ RolePO rolePO = getRolePOByMetalakeIdAndName(metalakeId,
identifier.name(), true);
List<SecurableObject> securableObjects = listSecurableObjects(rolePO);
@@ -447,14 +450,6 @@ public class RoleMetaService {
roleId, MetadataObject.Type.ROLE.name())));
}
- @Monitored(
- metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
- baseMetricName = "listSecurableObjectsByRoleId")
- public static List<SecurableObjectPO> listSecurableObjectsByRoleId(Long
roleId) {
- return SessionUtils.getWithoutCommit(
- SecurableObjectMapper.class, mapper ->
mapper.listSecurableObjectsByRoleId(roleId));
- }
-
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "listRolesByNamespace")
@@ -514,7 +509,8 @@ public class RoleMetaService {
}
private static List<SecurableObject> listSecurableObjects(RolePO po) {
- List<SecurableObjectPO> securableObjectPOs =
listSecurableObjectsByRoleId(po.getRoleId());
+ // Full role conversions must use privileges from the same statement as
the role metadata.
+ List<SecurableObjectPO> securableObjectPOs =
po.requireSecurableObjectPOs();
List<SecurableObject> securableObjects = Lists.newArrayList();
securableObjectPOs.stream()
@@ -555,10 +551,18 @@ public class RoleMetaService {
}
private static RolePO getRolePOByMetalakeIdAndName(Long metalakeId, String
roleName) {
+ return getRolePOByMetalakeIdAndName(metalakeId, roleName, false);
+ }
+
+ private static RolePO getRolePOByMetalakeIdAndName(
+ Long metalakeId, String roleName, boolean withPrivileges) {
RolePO rolePO =
SessionUtils.getWithoutCommit(
RoleMetaMapper.class,
- mapper -> mapper.selectRoleMetaByMetalakeIdAndName(metalakeId,
roleName));
+ mapper ->
+ withPrivileges
+ ?
mapper.selectRoleWithPrivilegesByMetalakeIdAndName(metalakeId, roleName)
+ : mapper.selectRoleMetaByMetalakeIdAndName(metalakeId,
roleName));
if (rolePO == null) {
throw new NoSuchEntityException(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
index 6f405747c6..f50680a537 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
@@ -63,6 +63,8 @@ public class TableMetaService {
private static final Logger LOG =
LoggerFactory.getLogger(TableMetaService.class);
private static final TableMetaService INSTANCE = new TableMetaService();
private BasePOStorageOps<TablePO, TableMetaMapper> ops;
+ private final BasePOStorageOps<TablePO, TableMetaMapper> snapshotOps =
+ new HierarchicalConversionPOStorageOps<>(new TablePOStorageOps(true));
public static TableMetaService getInstance() {
return INSTANCE;
@@ -124,13 +126,9 @@ public class TableMetaService {
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "getTableByIdentifier")
public TableEntity getTableByIdentifier(NameIdentifier identifier) {
- TablePO tablePO = getTablePOByIdentifier(identifier);
-
- List<ColumnPO> columnPOs =
- TableColumnMetaService.getInstance()
- .getColumnsByTableIdAndVersion(tablePO.getTableId(),
tablePO.getCurrentVersion());
-
- return POConverters.fromTableAndColumnPOs(tablePO, columnPOs,
identifier.namespace());
+ TablePO tablePO = getTablePOByIdentifier(identifier, snapshotOps);
+ return POConverters.fromTableAndColumnPOs(
+ tablePO, tablePO.requireColumnPOs(), identifier.namespace());
}
@Monitored(
@@ -237,10 +235,8 @@ public class TableMetaService {
@Monitored(metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "updateTable")
public <E extends Entity & HasIdentifier> TableEntity updateTable(
NameIdentifier identifier, Function<E, E> updater) throws IOException {
- TablePO oldTablePO = getTablePOByIdentifier(identifier);
- List<ColumnPO> oldTableColumns =
- TableColumnMetaService.getInstance()
- .getColumnsByTableIdAndVersion(oldTablePO.getTableId(),
oldTablePO.getCurrentVersion());
+ TablePO oldTablePO = getTablePOByIdentifier(identifier, snapshotOps);
+ List<ColumnPO> oldTableColumns = oldTablePO.requireColumnPOs();
TableEntity oldTableEntity =
POConverters.fromTableAndColumnPOs(oldTablePO, oldTableColumns,
identifier.namespace());
@@ -422,11 +418,17 @@ public class TableMetaService {
}
private TablePO getTablePOByIdentifier(NameIdentifier identifier) {
+ return getTablePOByIdentifier(identifier, ops);
+ }
+
+ private TablePO getTablePOByIdentifier(
+ NameIdentifier identifier, BasePOStorageOps<TablePO, TableMetaMapper>
readOps) {
NameIdentifierUtil.checkTable(identifier);
TablePO tablePO =
SessionUtils.getWithoutCommit(
TableMetaMapper.class,
- mapper -> POStorageReadRouting.getPO(mapper, identifier, ops,
Entity.EntityType.TABLE));
+ mapper ->
+ POStorageReadRouting.getPO(mapper, identifier, readOps,
Entity.EntityType.TABLE));
if (tablePO == null) {
throw new NoSuchEntityException(
NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TablePOStorageOps.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TablePOStorageOps.java
index 599a25fded..6bb336850c 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TablePOStorageOps.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TablePOStorageOps.java
@@ -29,7 +29,21 @@ import org.apache.gravitino.storage.relational.po.TablePO;
public class TablePOStorageOps extends BasePOStorageOps<TablePO,
TableMetaMapper> {
- public TablePOStorageOps() {}
+ private final boolean includeColumns;
+
+ public TablePOStorageOps() {
+ this(false);
+ }
+
+ /**
+ * Creates storage operations with optional single-statement column loading
for get operations.
+ * Lists remain metadata-only.
+ *
+ * @param includeColumns whether get operations load the column snapshot
+ */
+ public TablePOStorageOps(boolean includeColumns) {
+ this.includeColumns = includeColumns;
+ }
@Override
public void insertPO(TableMetaMapper mapper, TablePO tablePO, boolean
overwrite) {
@@ -47,15 +61,20 @@ public class TablePOStorageOps extends
BasePOStorageOps<TablePO, TableMetaMapper
@Override
public TablePO getPO(TableMetaMapper mapper, Long parentId, String name) {
- return mapper.selectTableMetaBySchemaIdAndName(parentId, name);
+ return includeColumns
+ ? mapper.selectTableWithColumnsBySchemaIdAndName(parentId, name)
+ : mapper.selectTableMetaBySchemaIdAndName(parentId, name);
}
@Override
public TablePO getPOByFullName(TableMetaMapper mapper, NameIdentifier
identifier) {
Namespace namespace = identifier.namespace();
TablePO po =
- mapper.selectTableByFullQualifiedName(
- namespace.level(0), namespace.level(1), namespace.level(2),
identifier.name());
+ includeColumns
+ ? mapper.selectTableWithColumnsByFullQualifiedName(
+ namespace.level(0), namespace.level(1), namespace.level(2),
identifier.name())
+ : mapper.selectTableByFullQualifiedName(
+ namespace.level(0), namespace.level(1), namespace.level(2),
identifier.name());
// INNER JOIN on metalake/catalog: a null PO means the metalake or catalog
does not exist.
if (po == null) {
throw new NoSuchEntityException(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/UserMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/UserMetaService.java
index b6c2fcd3f1..344adede9b 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/UserMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/UserMetaService.java
@@ -65,10 +65,17 @@ public class UserMetaService {
private UserMetaService() {}
private UserPO getUserPOByMetalakeIdAndName(Long metalakeId, String
userName) {
+ return getUserPOByMetalakeIdAndName(metalakeId, userName, false);
+ }
+
+ private UserPO getUserPOByMetalakeIdAndName(Long metalakeId, String
userName, boolean withRoles) {
UserPO userPO =
SessionUtils.getWithoutCommit(
UserMetaMapper.class,
- mapper -> mapper.selectUserMetaByMetalakeIdAndName(metalakeId,
userName));
+ mapper ->
+ withRoles
+ ?
mapper.selectUserWithRolesByMetalakeIdAndName(metalakeId, userName)
+ : mapper.selectUserMetaByMetalakeIdAndName(metalakeId,
userName));
if (userPO == null) {
throw new NoSuchEntityException(
@@ -105,8 +112,8 @@ public class UserMetaService {
Long metalakeId =
MetalakeMetaService.getInstance().getMetalakeIdByName(identifier.namespace().level(0));
- UserPO userPO = getUserPOByMetalakeIdAndName(metalakeId,
identifier.name());
- List<RolePO> rolePOs =
RoleMetaService.getInstance().listRolesByUserId(userPO.getUserId());
+ UserPO userPO = getUserPOByMetalakeIdAndName(metalakeId,
identifier.name(), true);
+ List<RolePO> rolePOs = userPO.requireRolePOs();
return POConverters.fromUserPO(userPO, rolePOs, identifier.namespace());
}
@@ -239,8 +246,8 @@ public class UserMetaService {
Long metalakeId =
MetalakeMetaService.getInstance().getMetalakeIdByName(identifier.namespace().level(0));
- UserPO oldUserPO = getUserPOByMetalakeIdAndName(metalakeId,
identifier.name());
- List<RolePO> rolePOs =
RoleMetaService.getInstance().listRolesByUserId(oldUserPO.getUserId());
+ UserPO oldUserPO = getUserPOByMetalakeIdAndName(metalakeId,
identifier.name(), true);
+ List<RolePO> rolePOs = oldUserPO.requireRolePOs();
UserEntity oldUserEntity = POConverters.fromUserPO(oldUserPO, rolePOs,
identifier.namespace());
UserEntity newEntity = (UserEntity) updater.apply((E) oldUserEntity);
diff --git
a/core/src/main/resources/org/apache/gravitino/storage/relational/mapper/RoleMetaMapper.xml
b/core/src/main/resources/org/apache/gravitino/storage/relational/mapper/RoleMetaMapper.xml
new file mode 100644
index 0000000000..fab177371e
--- /dev/null
+++
b/core/src/main/resources/org/apache/gravitino/storage/relational/mapper/RoleMetaMapper.xml
@@ -0,0 +1,43 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"https://mybatis.org/dtd/mybatis-3-mapper.dtd";>
+<mapper
namespace="org.apache.gravitino.storage.relational.mapper.RoleMetaMapper">
+ <resultMap id="rolePrivilegeSnapshot"
type="org.apache.gravitino.storage.relational.po.SecurableObjectPO">
+ <id property="id" column="relationId"/>
+ <result property="roleId" column="roleId"/>
+ <result property="metadataObjectId" column="metadataObjectId"/>
+ <result property="type" column="type"/>
+ <result property="privilegeNames" column="privilegeNames"/>
+ <result property="privilegeConditions" column="privilegeConditions"/>
+ <result property="currentVersion" column="currentVersion"/>
+ <result property="lastVersion" column="lastVersion"/>
+ <result property="deletedAt" column="deletedAt"/>
+ </resultMap>
+ <resultMap id="membershipRoleSnapshot"
type="org.apache.gravitino.storage.relational.po.RolePO">
+ <id property="roleId" column="roleId"/>
+ <result property="roleName" column="roleName"/>
+ <result property="metalakeId" column="metalakeId"/>
+ <result property="properties" column="properties"/>
+ <result property="auditInfo" column="auditInfo"/>
+ <result property="currentVersion" column="currentVersion"/>
+ <result property="lastVersion" column="lastVersion"/>
+ <result property="deletedAt" column="deletedAt"/>
+ </resultMap>
+</mapper>
diff --git
a/core/src/main/resources/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.xml
b/core/src/main/resources/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.xml
new file mode 100644
index 0000000000..d7f3ffac8a
--- /dev/null
+++
b/core/src/main/resources/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.xml
@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"https://mybatis.org/dtd/mybatis-3-mapper.dtd";>
+<mapper
namespace="org.apache.gravitino.storage.relational.mapper.TableMetaMapper">
+ <resultMap id="tableColumnSnapshot"
type="org.apache.gravitino.storage.relational.po.ColumnPO">
+ <id property="columnId" column="columnId"/>
+ <id property="tableVersion" column="tableVersion"/>
+ <result property="columnName" column="columnName"/>
+ <result property="columnPosition" column="columnPosition"/>
+ <result property="metalakeId" column="metalakeId"/>
+ <result property="catalogId" column="catalogId"/>
+ <result property="schemaId" column="schemaId"/>
+ <result property="tableId" column="tableId"/>
+ <result property="columnType" column="columnType"/>
+ <result property="columnComment" column="columnComment"/>
+ <result property="nullable" column="nullable"/>
+ <result property="autoIncrement" column="autoIncrement"/>
+ <result property="defaultValue" column="defaultValue"/>
+ <result property="columnOpType" column="columnOpType"/>
+ <result property="deletedAt" column="deletedAt"/>
+ <result property="auditInfo" column="auditInfo"/>
+ </resultMap>
+</mapper>
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestTableMetaBaseSQLProvider.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestTableMetaBaseSQLProvider.java
index a743e9108e..4bd55ccc73 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestTableMetaBaseSQLProvider.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestTableMetaBaseSQLProvider.java
@@ -63,4 +63,35 @@ class TestTableMetaBaseSQLProvider {
Assertions.assertTrue(sql.contains("WHERE table_id = #{tableId} AND
deleted_at = 0"));
Assertions.assertTrue(sql.endsWith("FOR UPDATE"));
}
+
+ @Test
+ void testAggregateReadRejectsDynamicRootSql() {
+ TableMetaBaseSQLProvider dynamic =
+ new TableMetaBaseSQLProvider() {
+ @Override
+ public String selectTableMetaBySchemaIdAndName(Long schemaId, String
tableName) {
+ return " <script>SELECT 1</script>";
+ }
+
+ @Override
+ public String selectTableByFullQualifiedName(
+ String metalake, String catalog, String schema, String table) {
+ return "<script>SELECT 1</script>";
+ }
+ };
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> dynamic.selectTableWithColumnsBySchemaIdAndName(1L, "table"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> dynamic.selectTableWithColumnsByFullQualifiedName("ml", "cat",
"schema", "table"));
+ }
+
+ @Test
+ void testAggregateReadAcceptsStaticRootWithTrailingSemicolon() {
+ String sql = PROVIDER.selectTableWithColumnsByFullQualifiedName("ml",
"cat", "schema", "table");
+ Assertions.assertTrue(sql.startsWith("SELECT t.*"));
+ Assertions.assertFalse(sql.contains(";)"));
+ Assertions.assertFalse(sql.contains("<script>"));
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestTablePO.java
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestTablePO.java
index 45db0357e1..f86dd5a7ed 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestTablePO.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestTablePO.java
@@ -22,6 +22,7 @@ import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.List;
+import org.apache.commons.lang3.reflect.FieldUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -48,6 +49,19 @@ public class TestTablePO {
}
Assertions.assertTrue(
dropped.isEmpty(), () -> "TablePO.builder(TablePO) does not copy these
fields: " + dropped);
+ Assertions.assertNotSame(source.requireColumnPOs(),
copy.requireColumnPOs());
+ copy.requireColumnPOs().clear();
+ Assertions.assertEquals(1, source.requireColumnPOs().size());
+ }
+
+ @Test
+ void testMetadataOnlyCopyRetainsUnloadedColumns() throws Exception {
+ TablePO source = fullyPopulated();
+ FieldUtils.writeField(source, "columnPOs", null, true);
+ TablePO copy = TablePO.builder(source).build();
+
+ Assertions.assertNull(copy.getColumnPOs());
+ Assertions.assertThrows(IllegalStateException.class,
copy::requireColumnPOs);
}
/** Every field set to a distinct value, so a missed copy cannot pass by
coincidence. */
@@ -71,6 +85,26 @@ public class TestTablePO {
.withIndexes("indexes")
.withComment("comment");
TablePO po = builder.build();
+ // Populate the aggregate field as MyBatis does, with a non-empty snapshot
so a missed copy
+ // fails.
+ ColumnPO column =
+ ColumnPO.builder()
+ .withColumnId(10L)
+ .withColumnName("column")
+ .withColumnPosition(0)
+ .withMetalakeId(2L)
+ .withCatalogId(3L)
+ .withSchemaId(4L)
+ .withTableId(1L)
+ .withTableVersion(5L)
+ .withColumnType("integer")
+ .withNullable((byte) 1)
+ .withAutoIncrement((byte) 0)
+ .withColumnOpType((byte) 0)
+ .withAuditInfo("audit")
+ .withDeletedAt(0L)
+ .build();
+ FieldUtils.writeField(po, "columnPOs", new ArrayList<>(List.of(column)),
true);
// Guard the guard: if a new field is added and left unset above, the
comparison would trivially
// hold with null on both sides and prove nothing.
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestConsistentEntityReads.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestConsistentEntityReads.java
new file mode 100644
index 0000000000..073343c52b
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestConsistentEntityReads.java
@@ -0,0 +1,774 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.service;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiConsumer;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.HasIdentifier;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.authorization.AuthorizationUtils;
+import org.apache.gravitino.authorization.Privileges;
+import org.apache.gravitino.authorization.SecurableObject;
+import org.apache.gravitino.authorization.SecurableObjects;
+import org.apache.gravitino.cache.CaffeineEntityCache;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.ColumnEntity;
+import org.apache.gravitino.meta.GroupEntity;
+import org.apache.gravitino.meta.RoleEntity;
+import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.meta.UserEntity;
+import org.apache.gravitino.rel.types.Types;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.storage.relational.RelationalEntityStore;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.GroupMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.UserMetaMapper;
+import org.apache.gravitino.storage.relational.po.GroupPO;
+import org.apache.gravitino.storage.relational.po.RolePO;
+import org.apache.gravitino.storage.relational.po.TablePO;
+import org.apache.gravitino.storage.relational.po.UserPO;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.junit.jupiter.api.TestTemplate;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+/** Exercises aggregate reads across a real concurrent commit on every JDBC
backend. */
+public class TestConsistentEntityReads extends TestJDBCBackend {
+ private static final String METALAKE = "consistent_reads";
+ private static final String CATALOG = "catalog";
+ private static final String SCHEMA = "schema";
+
+ @TestTemplate
+ void testTableOverwritePreservesSnapshot() throws Exception {
+ checkTableOverwrite(false, false);
+ }
+
+ @TestTemplate
+ void testTableReplacementPreservesSnapshot() throws Exception {
+ checkTableOverwrite(true, false);
+ }
+
+ @TestTemplate
+ void testCachedTableReadNeverPublishesEmptyColumns() throws Exception {
+ checkTableOverwrite(false, true);
+ }
+
+ @TestTemplate
+ void testTableDropPreservesSnapshot() throws Exception {
+ parents();
+ TableEntity original = table(RandomIdGenerator.INSTANCE.nextId(),
"before", "old_column");
+ backend.insert(original, false);
+ TableEntity read =
+ afterRootRead(
+ TableMetaMapper.class,
+ () -> backend.get(original.nameIdentifier(),
Entity.EntityType.TABLE),
+ () -> backend.delete(original.nameIdentifier(),
Entity.EntityType.TABLE, true));
+ assertTableSnapshot(original, read);
+ assertThrows(
+ NoSuchEntityException.class,
+ () -> backend.get(original.nameIdentifier(), Entity.EntityType.TABLE));
+ }
+
+ @TestTemplate
+ void testEmptyTableIsNotMistakenForMissing() throws Exception {
+ parents();
+ TableEntity empty =
+ TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("empty")
+ .withNamespace(Namespace.of(METALAKE, CATALOG, SCHEMA))
+ .withColumns(List.of())
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ backend.insert(empty, false);
+ TableEntity read = backend.get(empty.nameIdentifier(),
Entity.EntityType.TABLE);
+ assertEquals(empty.id(), read.id());
+ assertTrue(read.columns().isEmpty());
+ }
+
+ @TestTemplate
+ void testRoleOverwritePreservesPrivilegeSnapshot() throws Exception {
+ parents();
+ createAndInsertCatalog(METALAKE, "other_catalog");
+ RoleEntity before = role("before", CATALOG);
+ RoleEntity after = role("after", "other_catalog");
+ backend.insert(before, false);
+ RoleEntity read =
+ afterRootRead(
+ RoleMetaMapper.class,
+ () -> backend.get(before.nameIdentifier(), Entity.EntityType.ROLE),
+ () -> backend.insert(after, true));
+ assertEquals(before.properties(), read.properties());
+ assertEquals(1, read.securableObjects().size());
+ assertEquals(CATALOG, read.securableObjects().get(0).fullName());
+ RoleEntity current = backend.get(before.nameIdentifier(),
Entity.EntityType.ROLE);
+ assertEquals(after.properties(), current.properties());
+ assertEquals("other_catalog",
current.securableObjects().get(0).fullName());
+ }
+
+ @TestTemplate
+ void testRoleDropPreservesPrivilegeSnapshot() throws Exception {
+ parents();
+ RoleEntity before = role("before", CATALOG);
+ backend.insert(before, false);
+ RoleEntity read =
+ afterRootRead(
+ RoleMetaMapper.class,
+ () -> backend.get(before.nameIdentifier(), Entity.EntityType.ROLE),
+ () -> backend.delete(before.nameIdentifier(),
Entity.EntityType.ROLE, true));
+ assertEquals(1, read.securableObjects().size());
+ assertEquals(CATALOG, read.securableObjects().get(0).fullName());
+ assertThrows(
+ NoSuchEntityException.class,
+ () -> backend.get(before.nameIdentifier(), Entity.EntityType.ROLE));
+ }
+
+ @TestTemplate
+ void testRoleListingPreservesPrivilegeSnapshot() throws Exception {
+ parents();
+ createAndInsertCatalog(METALAKE, "other_catalog");
+ RoleEntity before = role("before", CATALOG);
+ RoleEntity after = role("after", "other_catalog");
+ backend.insert(before, false);
+ List<RoleEntity> read =
+ afterRootRead(
+ RoleMetaMapper.class,
+ () ->
+ RoleMetaService.getInstance()
+ .listRolesByMetadataObject(
+ NameIdentifier.of(METALAKE, CATALOG),
Entity.EntityType.CATALOG, true),
+ () -> backend.insert(after, true));
+ assertEquals(1, read.size());
+ assertEquals(before.properties(), read.get(0).properties());
+ assertEquals(CATALOG, read.get(0).securableObjects().get(0).fullName());
+ }
+
+ @TestTemplate
+ void testUserOverwritePreservesMembershipSnapshot() throws Exception {
+ parents();
+ RoleEntity first = role("before", CATALOG);
+ RoleEntity second =
+ createRoleEntity(12346L, first.namespace(), "second_role", AUDIT_INFO,
CATALOG);
+ backend.insert(first, false);
+ backend.insert(second, false);
+ UserEntity before =
+ createUserEntity(
+ 12347L,
+ AuthorizationUtils.ofUserNamespace(METALAKE),
+ "principal",
+ AUDIT_INFO,
+ List.of(first.name()),
+ List.of(first.id()));
+ AuditInfo afterAudit =
+
AuditInfo.builder().withCreator("after").withCreateTime(AUDIT_INFO.createTime()).build();
+ UserEntity after =
+ createUserEntity(
+ before.id(),
+ before.namespace(),
+ before.name(),
+ afterAudit,
+ List.of(second.name()),
+ List.of(second.id()));
+ backend.insert(before, false);
+ UserEntity read =
+ afterRootRead(
+ UserMetaMapper.class,
+ () -> backend.get(before.nameIdentifier(), Entity.EntityType.USER),
+ () -> backend.insert(after, true));
+ assertEquals(before.auditInfo(), read.auditInfo());
+ assertEquals(before.roleNames(), read.roleNames());
+ assertEquals(before.roleIds(), read.roleIds());
+ UserEntity current = backend.get(before.nameIdentifier(),
Entity.EntityType.USER);
+ assertEquals(after.auditInfo(), current.auditInfo());
+ assertEquals(after.roleIds(), current.roleIds());
+ }
+
+ @TestTemplate
+ void testUserDropPreservesMembershipSnapshot() throws Exception {
+ parents();
+ RoleEntity role = role("before", CATALOG);
+ backend.insert(role, false);
+ UserEntity before =
+ createUserEntity(
+ 12347L,
+ AuthorizationUtils.ofUserNamespace(METALAKE),
+ "principal",
+ AUDIT_INFO,
+ List.of(role.name()),
+ List.of(role.id()));
+ backend.insert(before, false);
+ UserEntity read =
+ afterRootRead(
+ UserMetaMapper.class,
+ () -> backend.get(before.nameIdentifier(), Entity.EntityType.USER),
+ () -> backend.delete(before.nameIdentifier(),
Entity.EntityType.USER, true));
+ assertEquals(before.roleNames(), read.roleNames());
+ assertEquals(before.roleIds(), read.roleIds());
+ assertThrows(
+ NoSuchEntityException.class,
+ () -> backend.get(before.nameIdentifier(), Entity.EntityType.USER));
+ }
+
+ @TestTemplate
+ void testGroupOverwritePreservesMembershipSnapshot() throws Exception {
+ parents();
+ RoleEntity first = role("before", CATALOG);
+ RoleEntity second =
+ createRoleEntity(12346L, first.namespace(), "second_role", AUDIT_INFO,
CATALOG);
+ backend.insert(first, false);
+ backend.insert(second, false);
+ GroupEntity before =
+ createGroupEntity(
+ 12347L,
+ AuthorizationUtils.ofGroupNamespace(METALAKE),
+ "principal",
+ AUDIT_INFO,
+ List.of(first.name()),
+ List.of(first.id()));
+ AuditInfo afterAudit =
+
AuditInfo.builder().withCreator("after").withCreateTime(AUDIT_INFO.createTime()).build();
+ GroupEntity after =
+ createGroupEntity(
+ before.id(),
+ before.namespace(),
+ before.name(),
+ afterAudit,
+ List.of(second.name()),
+ List.of(second.id()));
+ backend.insert(before, false);
+ GroupEntity read =
+ afterRootRead(
+ GroupMetaMapper.class,
+ () -> backend.get(before.nameIdentifier(),
Entity.EntityType.GROUP),
+ () -> backend.insert(after, true));
+ assertEquals(before.auditInfo(), read.auditInfo());
+ assertEquals(before.roleNames(), read.roleNames());
+ assertEquals(before.roleIds(), read.roleIds());
+ GroupEntity current = backend.get(before.nameIdentifier(),
Entity.EntityType.GROUP);
+ assertEquals(after.auditInfo(), current.auditInfo());
+ assertEquals(after.roleIds(), current.roleIds());
+ }
+
+ @TestTemplate
+ void testGroupDropPreservesMembershipSnapshot() throws Exception {
+ parents();
+ RoleEntity role = role("before", CATALOG);
+ backend.insert(role, false);
+ GroupEntity before =
+ createGroupEntity(
+ 12347L,
+ AuthorizationUtils.ofGroupNamespace(METALAKE),
+ "principal",
+ AUDIT_INFO,
+ List.of(role.name()),
+ List.of(role.id()));
+ backend.insert(before, false);
+ GroupEntity read =
+ afterRootRead(
+ GroupMetaMapper.class,
+ () -> backend.get(before.nameIdentifier(),
Entity.EntityType.GROUP),
+ () -> backend.delete(before.nameIdentifier(),
Entity.EntityType.GROUP, true));
+ assertEquals(before.roleNames(), read.roleNames());
+ assertEquals(before.roleIds(), read.roleIds());
+ assertThrows(
+ NoSuchEntityException.class,
+ () -> backend.get(before.nameIdentifier(), Entity.EntityType.GROUP));
+ }
+
+ @TestTemplate
+ void testFullNameTableReadPreservesSnapshot() throws Exception {
+ GravitinoEnv environment = Mockito.spy(GravitinoEnv.getInstance());
+ Mockito.doReturn(false).when(environment).cacheEnabled();
+ try (MockedStatic<GravitinoEnv> environments =
Mockito.mockStatic(GravitinoEnv.class)) {
+ environments.when(GravitinoEnv::getInstance).thenReturn(environment);
+ checkTableOverwrite(false, false);
+ }
+ }
+
+ @TestTemplate
+ void testFullNameReadRetainsMissingEntityErrors() throws Exception {
+ parents();
+ GravitinoEnv environment = Mockito.spy(GravitinoEnv.getInstance());
+ Mockito.doReturn(false).when(environment).cacheEnabled();
+ try (MockedStatic<GravitinoEnv> environments =
Mockito.mockStatic(GravitinoEnv.class)) {
+ environments.when(GravitinoEnv::getInstance).thenReturn(environment);
+ for (NameIdentifier missing :
+ List.of(
+ NameIdentifier.of(METALAKE, CATALOG, SCHEMA, "absent"),
+ NameIdentifier.of(METALAKE, CATALOG, "absent_schema", "absent"),
+ NameIdentifier.of(METALAKE, "absent_catalog", SCHEMA,
"absent"))) {
+ assertThrows(
+ NoSuchEntityException.class, () -> backend.get(missing,
Entity.EntityType.TABLE));
+ }
+ }
+ }
+
+ @TestTemplate
+ void testRolePreservesDistinctRowsForSameTarget() throws Exception {
+ parents();
+ List<SecurableObject> objects =
+ List.of(
+ SecurableObjects.ofCatalog(CATALOG,
List.of(Privileges.UseCatalog.allow())),
+ SecurableObjects.ofCatalog(CATALOG,
List.of(Privileges.CreateSchema.allow())));
+ RoleEntity role =
+ createRoleEntity(
+ 12345L,
+ AuthorizationUtils.ofRoleNamespace(METALAKE),
+ "duplicate_targets",
+ AUDIT_INFO,
+ objects,
+ Map.of());
+ backend.insert(role, false);
+ RoleEntity read = backend.get(role.nameIdentifier(),
Entity.EntityType.ROLE);
+ assertEquals(2, read.securableObjects().size());
+ assertTrue(read.securableObjects().containsAll(objects));
+ List<RoleEntity> roles =
+ RoleMetaService.getInstance()
+ .listRolesByMetadataObject(
+ NameIdentifier.of(METALAKE, CATALOG),
Entity.EntityType.CATALOG, true);
+ assertEquals(1, roles.size());
+ assertEquals(2, roles.get(0).securableObjects().size());
+ assertTrue(roles.get(0).securableObjects().containsAll(objects));
+ backend.update(
+ role.nameIdentifier(),
+ Entity.EntityType.ROLE,
+ current -> {
+ RoleEntity snapshot = (RoleEntity) current;
+ assertEquals(2, snapshot.securableObjects().size());
+ assertTrue(snapshot.securableObjects().containsAll(objects));
+ return snapshot;
+ });
+ }
+
+ @TestTemplate
+ void testEmptyRoleRetainsEmptyPrivilegeSnapshot() throws Exception {
+ parents();
+ RoleEntity replacement = role("after", CATALOG);
+ RoleEntity empty =
+ RoleEntity.builder()
+ .withId(replacement.id())
+ .withName(replacement.name())
+ .withNamespace(replacement.namespace())
+ .withProperties(Map.of("marker", "before"))
+ .withSecurableObjects(List.of())
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ backend.insert(empty, false);
+ RoleEntity read =
+ afterRootRead(
+ RoleMetaMapper.class,
+ () -> backend.get(empty.nameIdentifier(), Entity.EntityType.ROLE),
+ () -> backend.insert(replacement, true));
+ assertEquals(empty.properties(), read.properties());
+ assertTrue(read.securableObjects().isEmpty());
+ RoleEntity current = backend.get(empty.nameIdentifier(),
Entity.EntityType.ROLE);
+ assertEquals(replacement.securableObjects(), current.securableObjects());
+ }
+
+ @TestTemplate
+ void testTableUpdaterReceivesConsistentSnapshot() throws Exception {
+ parents();
+ TableEntity before = table(RandomIdGenerator.INSTANCE.nextId(), "before",
"old_column");
+ TableEntity after = table(before.id(), "after", "new_column");
+ assertUpdaterSnapshot(TableMetaMapper.class, before, after,
this::assertTableSnapshot);
+ }
+
+ @TestTemplate
+ void testRoleUpdaterReceivesConsistentSnapshot() throws Exception {
+ parents();
+ createAndInsertCatalog(METALAKE, "other_catalog");
+ assertUpdaterSnapshot(
+ RoleMetaMapper.class,
+ role("before", CATALOG),
+ role("after", "other_catalog"),
+ (expected, actual) -> {
+ assertEquals(expected.properties(), actual.properties());
+ assertEquals(expected.securableObjects(), actual.securableObjects());
+ });
+ }
+
+ @TestTemplate
+ void testUserUpdaterReceivesConsistentSnapshot() throws Exception {
+ parents();
+ RoleEntity first = role("before", CATALOG);
+ RoleEntity second =
+ createRoleEntity(12346L, first.namespace(), "second_role", AUDIT_INFO,
CATALOG);
+ backend.insert(first, false);
+ backend.insert(second, false);
+ Namespace namespace = AuthorizationUtils.ofUserNamespace(METALAKE);
+ UserEntity before =
+ createUserEntity(
+ 12347L, namespace, "principal", AUDIT_INFO, List.of(first.name()),
List.of(first.id()));
+ UserEntity after =
+ createUserEntity(
+ before.id(),
+ namespace,
+ before.name(),
+ AUDIT_INFO,
+ List.of(second.name()),
+ List.of(second.id()));
+ assertUpdaterSnapshot(
+ UserMetaMapper.class,
+ before,
+ after,
+ (expected, actual) -> {
+ assertEquals(expected.auditInfo(), actual.auditInfo());
+ assertEquals(expected.roleNames(), actual.roleNames());
+ assertEquals(expected.roleIds(), actual.roleIds());
+ });
+ }
+
+ @TestTemplate
+ void testGroupUpdaterReceivesConsistentSnapshot() throws Exception {
+ parents();
+ RoleEntity first = role("before", CATALOG);
+ RoleEntity second =
+ createRoleEntity(12346L, first.namespace(), "second_role", AUDIT_INFO,
CATALOG);
+ backend.insert(first, false);
+ backend.insert(second, false);
+ Namespace namespace = AuthorizationUtils.ofGroupNamespace(METALAKE);
+ GroupEntity before =
+ createGroupEntity(
+ 12347L, namespace, "principal", AUDIT_INFO, List.of(first.name()),
List.of(first.id()));
+ GroupEntity after =
+ createGroupEntity(
+ before.id(),
+ namespace,
+ before.name(),
+ AUDIT_INFO,
+ List.of(second.name()),
+ List.of(second.id()));
+ assertUpdaterSnapshot(
+ GroupMetaMapper.class,
+ before,
+ after,
+ (expected, actual) -> {
+ assertEquals(expected.auditInfo(), actual.auditInfo());
+ assertEquals(expected.roleNames(), actual.roleNames());
+ assertEquals(expected.roleIds(), actual.roleIds());
+ });
+ }
+
+ @TestTemplate
+ void testMetadataOnlyReadsDoNotMasqueradeAsEmptySnapshots() throws Exception
{
+ parents();
+ TableEntity table = table(RandomIdGenerator.INSTANCE.nextId(), "before",
"column");
+ RoleEntity role = role("before", CATALOG);
+ UserEntity user =
+ createUserEntity(
+ 12347L,
+ AuthorizationUtils.ofUserNamespace(METALAKE),
+ "user",
+ AUDIT_INFO,
+ List.of(),
+ List.of());
+ GroupEntity group =
+ createGroupEntity(
+ 12348L,
+ AuthorizationUtils.ofGroupNamespace(METALAKE),
+ "group",
+ AUDIT_INFO,
+ List.of(),
+ List.of());
+ backend.insert(table, false);
+ backend.insert(role, false);
+ backend.insert(user, false);
+ backend.insert(group, false);
+ long metalakeId =
+ EntityIdService.getEntityId(NameIdentifier.of(METALAKE),
Entity.EntityType.METALAKE);
+ long schemaId =
+ EntityIdService.getEntityId(
+ NameIdentifier.of(METALAKE, CATALOG, SCHEMA),
Entity.EntityType.SCHEMA);
+ TablePO tablePO =
+ SessionUtils.getWithoutCommit(
+ TableMetaMapper.class,
+ mapper -> mapper.selectTableMetaBySchemaIdAndName(schemaId,
table.name()));
+ RolePO rolePO =
+ SessionUtils.getWithoutCommit(
+ RoleMetaMapper.class,
+ mapper -> mapper.selectRoleMetaByMetalakeIdAndName(metalakeId,
role.name()));
+ UserPO userPO =
+ SessionUtils.getWithoutCommit(
+ UserMetaMapper.class,
+ mapper -> mapper.selectUserMetaByMetalakeIdAndName(metalakeId,
user.name()));
+ GroupPO groupPO =
+ SessionUtils.getWithoutCommit(
+ GroupMetaMapper.class,
+ mapper -> mapper.selectGroupMetaByMetalakeIdAndName(metalakeId,
group.name()));
+ assertNull(tablePO.getColumnPOs());
+ assertNull(rolePO.getSecurableObjectPOs());
+ assertNull(userPO.getRolePOs());
+ assertNull(groupPO.getRolePOs());
+ assertThrows(IllegalStateException.class, tablePO::requireColumnPOs);
+ assertThrows(IllegalStateException.class,
rolePO::requireSecurableObjectPOs);
+ assertThrows(IllegalStateException.class, userPO::requireRolePOs);
+ assertThrows(IllegalStateException.class, groupPO::requireRolePOs);
+ assertMetadataOnlySnapshotRejected(TableMetaMapper.class, tablePO, table);
+ assertMetadataOnlySnapshotRejected(RoleMetaMapper.class, rolePO, role);
+ assertMetadataOnlySnapshotRejected(UserMetaMapper.class, userPO, user);
+ assertMetadataOnlySnapshotRejected(GroupMetaMapper.class, groupPO, group);
+ // A real aggregate mapper must initialize an empty collection even
without a joined child.
+ UserPO loadedUser =
+ SessionUtils.getWithoutCommit(
+ UserMetaMapper.class,
+ mapper ->
mapper.selectUserWithRolesByMetalakeIdAndName(metalakeId, user.name()));
+ GroupPO loadedGroup =
+ SessionUtils.getWithoutCommit(
+ GroupMetaMapper.class,
+ mapper ->
mapper.selectGroupWithRolesByMetalakeIdAndName(metalakeId, group.name()));
+ assertTrue(loadedUser.requireRolePOs().isEmpty());
+ assertTrue(loadedGroup.requireRolePOs().isEmpty());
+ }
+
+ private <E extends Entity & HasIdentifier> void
assertMetadataOnlySnapshotRejected(
+ Class<?> mapper, Object metadataOnly, E entity) throws Exception {
+ AtomicBoolean updaterInvoked = new AtomicBoolean();
+ try (MockedStatic<SessionUtils> ignored =
+ Mockito.mockStatic(
+ SessionUtils.class,
+ invocation -> {
+ Object result = invocation.callRealMethod();
+ if (invocation.getMethod().getName().equals("getWithoutCommit")
+ && invocation.getArgument(0) == mapper
+ && result != null
+ && result.getClass() == metadataOnly.getClass()) {
+ return metadataOnly;
+ }
+ return result;
+ })) {
+ assertThrows(
+ IllegalStateException.class, () ->
backend.get(entity.nameIdentifier(), entity.type()));
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ backend.<E>update(
+ entity.nameIdentifier(),
+ entity.type(),
+ current -> {
+ updaterInvoked.set(true);
+ return current;
+ }));
+ assertFalse(updaterInvoked.get(), "metadata-only input must be rejected
before the updater");
+ }
+ }
+
+ private <E extends Entity & HasIdentifier> void assertUpdaterSnapshot(
+ Class<?> mapper, E before, E after, BiConsumer<E, E> assertSnapshot)
throws Exception {
+ backend.insert(before, false);
+ AtomicBoolean invoked = new AtomicBoolean();
+ afterRootRead(
+ mapper,
+ () ->
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ backend.<E>update(
+ before.nameIdentifier(),
+ before.type(),
+ current -> {
+ assertSnapshot.accept(before, current);
+ invoked.set(true);
+ return current;
+ })),
+ () -> backend.insert(after, true));
+ assertTrue(invoked.get(), "the updater must receive the observed snapshot
before CAS fails");
+ assertSnapshot.accept(after, backend.get(after.nameIdentifier(),
after.type()));
+ }
+
+ private void checkTableOverwrite(boolean replaceId, boolean cached) throws
Exception {
+ parents();
+ TableEntity original = table(RandomIdGenerator.INSTANCE.nextId(),
"before", "old_column");
+ TableEntity replacement =
+ table(
+ replaceId ? RandomIdGenerator.INSTANCE.nextId() : original.id(),
"after", "new_column");
+ backend.insert(original, false);
+ RelationalEntityStore store = new RelationalEntityStore();
+ CaffeineEntityCache cache = new CaffeineEntityCache(new Config(false) {});
+ try {
+ FieldUtils.writeField(store, "backend", backend, true);
+ FieldUtils.writeField(store, "cache", cache, true);
+ TableEntity read;
+ read =
+ afterRootRead(
+ TableMetaMapper.class,
+ () -> {
+ try {
+ return cached
+ ? store.get(
+ original.nameIdentifier(), Entity.EntityType.TABLE,
TableEntity.class)
+ : backend.get(original.nameIdentifier(),
Entity.EntityType.TABLE);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ },
+ () -> backend.insert(replacement, true));
+ assertTableSnapshot(original, read);
+ if (cached) {
+ assertTableSnapshot(
+ original,
+ store.get(original.nameIdentifier(), Entity.EntityType.TABLE,
TableEntity.class));
+ }
+ TableEntity current = backend.get(original.nameIdentifier(),
Entity.EntityType.TABLE);
+ assertTableSnapshot(replacement, current);
+ } finally {
+ cache.clear();
+ }
+ }
+
+ private void parents() throws Exception {
+ createAndInsertMakeLake(METALAKE);
+ createAndInsertCatalog(METALAKE, CATALOG);
+ createAndInsertSchema(METALAKE, CATALOG, SCHEMA);
+ }
+
+ private TableEntity table(long id, String comment, String columnName) {
+ ColumnEntity column =
+ ColumnEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(columnName)
+ .withPosition(0)
+ .withDataType(Types.IntegerType.get())
+ .withNullable(true)
+ .withAutoIncrement(false)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ return TableEntity.builder()
+ .withId(id)
+ .withName("table")
+ .withNamespace(Namespace.of(METALAKE, CATALOG, SCHEMA))
+ .withColumns(
+ List.of(
+ column,
+ ColumnEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(columnName + "_second")
+ .withPosition(1)
+ .withDataType(Types.StringType.get())
+ .withNullable(false)
+ .withAutoIncrement(false)
+ .withAuditInfo(AUDIT_INFO)
+ .build()))
+ .withComment(comment)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
+ private RoleEntity role(String marker, String catalog) {
+ RoleEntity role =
+ createRoleEntity(
+ 12345L, AuthorizationUtils.ofRoleNamespace(METALAKE), "role",
AUDIT_INFO, catalog);
+ return RoleEntity.builder()
+ .withId(role.id())
+ .withName(role.name())
+ .withNamespace(role.namespace())
+ .withProperties(Map.of("marker", marker))
+ .withSecurableObjects(role.securableObjects())
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
+ private void assertTableSnapshot(TableEntity expected, TableEntity actual) {
+ assertEquals(expected.id(), actual.id());
+ assertEquals(expected.comment(), actual.comment());
+ assertEquals(expected.columns().size(), actual.columns().size());
+ Map<Long, ColumnEntity> actualById = new HashMap<>();
+ actual.columns().forEach(column -> actualById.put(column.id(), column));
+ assertEquals(expected.columns().size(), actualById.size());
+ for (ColumnEntity column : expected.columns()) {
+ ColumnEntity observed = actualById.get(column.id());
+ assertEquals(column.name(), observed.name());
+ assertEquals(column.position(), observed.position());
+ assertEquals(column.dataType(), observed.dataType());
+ assertEquals(column.nullable(), observed.nullable());
+ }
+ }
+
+ private <T> T afterRootRead(Class<?> rootMapper, CheckedRead<T> read,
CheckedWrite write)
+ throws Exception {
+ ExecutorService writer = Executors.newSingleThreadExecutor();
+ AtomicBoolean committed = new AtomicBoolean();
+ try (MockedStatic<SessionUtils> ignored =
+ Mockito.mockStatic(
+ SessionUtils.class,
+ invocation -> {
+ Object result = invocation.callRealMethod();
+ if (invocation.getMethod().getName().equals("getWithoutCommit")
+ && invocation.getArgument(0) == rootMapper
+ && (result instanceof TablePO
+ || result instanceof UserPO
+ || result instanceof GroupPO
+ || result instanceof RolePO
+ || result instanceof List
+ && !((List<?>) result).isEmpty()
+ && ((List<?>) result).get(0) instanceof RolePO)
+ && committed.compareAndSet(false, true)) {
+ // Commit on another connection after the root SELECT, before
any subsequent read.
+ // The joined read already owns all its dependent rows.
+ writer
+ .submit(
+ () -> {
+ write.run();
+ return null;
+ })
+ .get(30, TimeUnit.SECONDS);
+ }
+ return result;
+ })) {
+ T result = read.get();
+ assertTrue(committed.get(), "the writer must commit at the observed read
boundary");
+ return result;
+ } finally {
+ writer.shutdownNow();
+ assertTrue(writer.awaitTermination(30, TimeUnit.SECONDS));
+ }
+ }
+
+ @FunctionalInterface
+ private interface CheckedRead<T> {
+ T get() throws Exception;
+ }
+
+ @FunctionalInterface
+ private interface CheckedWrite {
+ void run() throws Exception;
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
index e140139045..a7fbd048bd 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
@@ -77,6 +77,7 @@ import
org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
import org.apache.gravitino.storage.relational.mapper.GroupMetaMapper;
import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
import org.apache.gravitino.storage.relational.mapper.TagMetaMapper;
import org.apache.gravitino.storage.relational.mapper.UserMetaMapper;
import org.apache.gravitino.storage.relational.po.CatalogPO;
@@ -188,7 +189,11 @@ class TestRoleMetaService extends TestJDBCBackend {
// The failed grant must roll back the role CAS and properties. Deletion
must still remove
// the old relation, and no replacement relation may survive for the
deleted endpoint.
assertEquals(beforeUpdate, getRolePO(role.name()));
-
assertTrue(RoleMetaService.listSecurableObjectsByRoleId(role.id()).isEmpty());
+ assertTrue(
+ SessionUtils.getWithoutCommit(
+ SecurableObjectMapper.class,
+ mapper -> mapper.listSecurableObjectsByRoleId(role.id()))
+ .isEmpty());
assertTrue(service.getRoleByIdentifier(role.nameIdentifier()).securableObjects().isEmpty());
Assertions.assertThrows(
NoSuchEntityException.class,
diff --git a/design-docs/entity-store-consistent-reads-audit.md
b/design-docs/entity-store-consistent-reads-audit.md
new file mode 100644
index 0000000000..d65d78eacf
--- /dev/null
+++ b/design-docs/entity-store-consistent-reads-audit.md
@@ -0,0 +1,71 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Entity-store aggregate read consistency audit
+
+Related issue: #13853, table-read P0. This audit covers entity-store reads,
not consistency with external catalogs or authorization plugins.
+
+## Invariant
+
+An aggregate returned by an entity-store get must describe one committed
statement snapshot: table metadata and its columns, role metadata and its
stored privileges, or principal metadata and its memberships. A concurrent
overwrite/drop may leave the reader with the complete older snapshot or with a
not-found result; it must not produce a root from one state and dependent rows
from another. This also applies to the old entity supplied to an updater.
+
+The reads use ordinary, non-locking SELECTs at the existing transaction
isolation level. No retry, timeout, new HTTP failure signal, migration, or
change to write CAS is introduced. The standard entity-cache population path
receives an already-complete table. Cross-node invalidation remains eventual:
this change prevents torn snapshots, not the normal delay before a peer replays
a committed change log.
+
+## Read audit
+
+| Read | Before | Result |
+| --- | --- | --- |
+| Table get and updater input | Root/current-version join followed by a
separate column-history query | One SELECT joins the root, current version, and
latest live history row for every column. Both parent-ID and full-name routes
are covered. |
+| Table list and batch get | Metadata-only, without columns | Unchanged; these
APIs deliberately return metadata-only entities. |
+| Role get, updater input, and all-fields list by metadata object | Role row
followed by privilege rows | One SELECT joins the role and all live stored
privileges. Metadata-only lists remain metadata-only. |
+| User and group get and updater input | Principal row followed by
membership/role rows | One SELECT joins the principal, live memberships, and
live roles. Empty memberships retain the existing representation. |
+| User/role batch get | Calls the single-entity getter once per identifier |
Each entity now benefits from its joined snapshot. There is no snapshot
covering the entire batch. |
+| Group batch get; user/group namespace and paginated list | Joined/aggregated
principal and membership queries | Each payload query already uses one SQL
statement; unchanged. |
+| Fileset | Joined root and location/version rows | Already one statement
through the nested result map; unchanged. |
+| Function, view, semantic model | Joined root and current-version rows;
structured content serialized in the version row | Already one statement;
unchanged. |
+| Metalake, catalog, schema, topic, model, tag, policy, job, job template |
Entity content comes from one root/version query | No separate dependent-row
assembly found in their single-entity getters; unchanged. Parent-name/ID
resolution can precede the read, but does not splice two versions of an entity
payload. |
+| Model version | Version rows and aliases protected by aggregate-version
revalidation | Existing bounded stable-read loop remains. Conflict rates during
linking are the separate P1 in #13853. |
+
+Role privilege target names are still resolved by stable metadata IDs after
reading the role/privilege snapshot. A target deleted before name resolution is
omitted, with the existing warning; target renames and the parent-name reads
are not a global snapshot of the hierarchy. This patch does not change that
behavior or promise cross-entity snapshot isolation.
+
+Relation lists that first resolve a principal/target and then query its
relations retain their existing list semantics. They do not assemble a full
root-plus-dependent payload for the resolved object. They are not converted
into transactions by this patch.
+
+## SQL and cost
+
+Table columns are versioned deltas, not complete per-version snapshots. The
joined query groups live history rows by `(table_id, column_id)` and takes the
maximum version no greater than the root's current version. DELETE markers must
participate in this selection; the final join excludes them only after
resolving the latest version. LEFT JOINs preserve tables with zero columns and
full-name missing-parent placeholders.
+
+History is aggregated once, rather than scanned in a correlated subquery for
every output column. Root selection appears twice inside the single statement
so both occurrences have explicitly bound parameters across supported backends.
The final history-row lookup uses the existing `(table_id, table_version,
column_id, deleted_at)` unique key in H2, MySQL and PostgreSQL. Role privilege
rows are grouped by their physical `id` primary key in the nested mapping. The
database does not require [...]
+
+The full-name route repeats the metalake/catalog/schema/table/version root
joins, including when the entity-ID cache is disabled. This is a deliberate
portability tradeoff: both derived tables use the same statically bound root
SQL and one statement snapshot without backend-specific LATERAL syntax. The
history aggregation remains bounded to the selected table. The earlier EXPLAIN
checks establish history/index access, not a throughput result for this
full-name route; this review round do [...]
+
+Aggregate PO child collections are null on metadata-only reads. MyBatis
initializes the nested collection, including a genuinely empty collection, only
for an aggregate result map. Full entity conversions and updater inputs require
that loaded collection and reject a metadata-only PO before computing privilege
or membership changes. The raw getters remain nullable for MyBatis mapping;
consumers use the checked `require...` accessors. `TablePO.builder(source)`
retains the loaded/unloaded [...]
+
+The nested result maps live in mapper XML resources, without public
placeholder query methods. Table aggregate wrapping accepts only static root
SQL; a MyBatis `<script>` root is rejected before executing SQL. One trailing
statement terminator is removed before nesting the root.
+
+Role reads no longer call the separate
`RoleMetaService.listSecurableObjectsByRoleId` service method. That test-only
service entry point is removed; the raw mapper remains for targeted storage
assertions. Its former `listSecurableObjectsByRoleId` service metric is no
longer emitted. Monitoring should use the enclosing role-read operations rather
than treating that metric's absence as a healthy authorization lookup rate.
+
+Root/name predicates and soft-delete filtering are inherited from the existing
providers, including backend overrides.
+
+Joined results repeat root fields for each child row, increasing result bytes
for wide tables or large properties. This trades additional root projection
bytes for one consistent read and fewer round trips; no throughput improvement
is claimed. Row-only identity and delete lookups retain their existing smaller
projection.
+
+## Regression verification
+
+`TestConsistentEntityReads` commits a write on another connection after the
reader's root-mapper call completes and before its next statement. Real
mapper/service/backend calls run on both threads; the mock only controls the
statement boundary. With the former split read this deterministically loses
columns, changes privileges, or mixes memberships. With the joined read, all
dependent rows were fetched before that boundary.
+
+Coverage includes same-ID overwrite, replacement by another ID, drop, a cache
miss followed by a cache hit, genuinely empty tables, multi-column snapshots,
metadata-only versus loaded-empty guards, parent-ID/full-name table routing,
missing-parent errors, role overwrite/drop/all-fields listing, empty privilege
snapshots, multiple privilege rows for the same target, and user/group
overwrite/drop. For table, role, user and group updates, a concurrent overwrite
commits after the root read; [...]
