[#10770] feat(auth): Add group cache mapper support (#11014)

### What changes were proposed in this pull request?

This PR splits the group-related authorization cache mapper work from
#10996.

It adds:
- `group_meta.updated_at` to H2/MySQL/PostgreSQL 1.3.0 schemas and
1.2.0-to-1.3.0 upgrade scripts.
- Group updated-at mapper APIs and SQL providers.
- `GroupUpdatedAt` PO.
- `GroupMetaService.updateGroup` updates `group_meta.updated_at` when
group role assignments change.
- Mapper and service tests for group updated-at behavior.

### Why are the changes needed?

Group role cache validation needs a DB-side version sentinel, matching
the existing user/role cache pattern.

Fix: #10770

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

```bash
./gradlew :core:spotlessApply
./gradlew :core:test --tests org.apache.gravitino.storage.relational.mapper.provider.base.TestAuthMappers --tests org.apache.gravitino.storage.relational.service.TestGroupMetaService
```

Author: yuqi1129

core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java

@@ -22,6 +22,7 @@
 import java.util.List;
 import org.apache.gravitino.storage.relational.po.ExtendedGroupPO;
 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.Param;
@@ -88,4 +89,11 @@ Integer updateGroupMeta(
       method = "deleteGroupMetasByLegacyTimeline")
   Integer deleteGroupMetasByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit);
+
+  @UpdateProvider(type = GroupMetaSQLProviderFactory.class, method = "touchGroupUpdatedAt")
+  void touchGroupUpdatedAt(@Param("groupId") long groupId);
+
+  @SelectProvider(type = GroupMetaSQLProviderFactory.class, method = "getGroupUpdatedAt")
+  GroupUpdatedAt getGroupUpdatedAt(
+      @Param("metalakeName") String metalakeName, @Param("groupName") String groupName);
 }

core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java

@@ -95,4 +95,13 @@ public static String deleteGroupMetasByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit) {
     return getProvider().deleteGroupMetasByLegacyTimeline(legacyTimeline, limit);
   }
+
+  public static String touchGroupUpdatedAt(@Param("groupId") long groupId) {
+    return getProvider().touchGroupUpdatedAt(groupId);
+  }
+
+  public static String getGroupUpdatedAt(
+      @Param("metalakeName") String metalakeName, @Param("groupName") String groupName) {
+    return getProvider().getGroupUpdatedAt(metalakeName, groupName);
+  }
 }

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java

@@ -182,4 +182,25 @@ public String deleteGroupMetasByLegacyTimeline(
         + GROUP_TABLE_NAME
         + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT #{limit}";
   }
+
+  public String touchGroupUpdatedAt(@Param("groupId") long groupId) {
+    return "UPDATE "
+        + GROUP_TABLE_NAME
+        + " SET updated_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE group_id = #{groupId} AND deleted_at = 0";
+  }
+
+  public String getGroupUpdatedAt(
+      @Param("metalakeName") String metalakeName, @Param("groupName") String groupName) {
+    return "SELECT gm.group_id as groupId, gm.updated_at as updatedAt"
+        + " FROM "
+        + GROUP_TABLE_NAME
+        + " gm"
+        + " JOIN "
+        + MetalakeMetaMapper.TABLE_NAME
+        + " mm ON gm.metalake_id = mm.metalake_id AND mm.deleted_at = 0"
+        + " WHERE mm.metalake_name = #{metalakeName} AND gm.group_name = #{groupName}"
+        + " AND gm.deleted_at = 0";
+  }
 }

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/GroupMetaPostgreSQLProvider.java

@@ -104,4 +104,12 @@ public String deleteGroupMetasByLegacyTimeline(
         + GROUP_TABLE_NAME
         + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT #{limit})";
   }
+
+  @Override
+  public String touchGroupUpdatedAt(@Param("groupId") long groupId) {
+    return "UPDATE "
+        + GROUP_TABLE_NAME
+        + " SET updated_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)"
+        + " WHERE group_id = #{groupId} AND deleted_at = 0";
+  }
 }

core/src/main/java/org/apache/gravitino/storage/relational/po/auth/GroupUpdatedAt.java

@@ -0,0 +1,34 @@
+/*
+ * 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.po.auth;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+/** Group identity + role-list staleness sentinel. */
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+public class GroupUpdatedAt {
+  private long groupId;
+  private long updatedAt;
+}

core/src/main/java/org/apache/gravitino/storage/relational/service/GroupMetaService.java

@@ -259,7 +259,11 @@ public <E extends Entity & HasIdentifier> GroupEntity updateGroup(
                 mapper ->
                     mapper.softDeleteGroupRoleRelByGroupAndRoles(
                         newEntity.id(), Lists.newArrayList(deleteRoleIds)));
-          });
+          },
+          () ->
+              SessionUtils.doWithoutCommit(
+                  GroupMetaMapper.class,
+                  mapper -> mapper.touchGroupUpdatedAt(oldGroupPO.getGroupId())));
     } catch (RuntimeException re) {
       ExceptionUtils.checkSQLException(
           re, Entity.EntityType.GROUP, newEntity.nameIdentifier().toString());

core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestAuthMappers.java

@@ -47,6 +47,7 @@
 import org.apache.gravitino.storage.relational.po.RolePO;
 import org.apache.gravitino.storage.relational.po.UserPO;
 import org.apache.gravitino.storage.relational.po.auth.ChangedOwnerInfo;
+import org.apache.gravitino.storage.relational.po.auth.GroupUpdatedAt;
 import org.apache.gravitino.storage.relational.po.auth.OwnerInfo;
 import org.apache.gravitino.storage.relational.po.auth.RoleUpdatedAt;
 import org.apache.gravitino.storage.relational.po.auth.UserUpdatedAt;
@@ -241,6 +242,51 @@ void testUserMetaGetUserUpdatedAt() {
     Assertions.assertEquals(expected, info.getUpdatedAt());
   }
 
+  @Test
+  void testGroupMetaTouchUpdatedAt() {
+    insertMetalake(1L, "metalake1");
+    insertGroup(30L, "group30", 1L);
+
+    long before = queryUpdatedAt("group_meta", "group_id", 30L);
+    Assertions.assertEquals(0L, before);
+
+    long jvmBefore = System.currentTimeMillis();
+    groupMetaMapper.touchGroupUpdatedAt(30L);
+    long jvmAfter = System.currentTimeMillis();
+
+    long after = queryUpdatedAt("group_meta", "group_id", 30L);
+    Assertions.assertTrue(
+        after >= jvmBefore - 1000L && after <= jvmAfter + 1000L,
+        "expected DB-time updated_at within 1s of JVM clock, got " + after);
+  }
+
+  @Test
+  void testGroupMetaTouchUpdatedAtSkipsSoftDeleted() {
+    insertMetalake(1L, "metalake1");
+    insertGroup(31L, "group31", 1L);
+    groupMetaMapper.softDeleteGroupMetaByGroupId(31L);
+
+    long beforeUpdatedAt = queryUpdatedAt("group_meta", "group_id", 31L);
+    groupMetaMapper.touchGroupUpdatedAt(31L);
+    long afterUpdatedAt = queryUpdatedAt("group_meta", "group_id", 31L);
+
+    Assertions.assertEquals(beforeUpdatedAt, afterUpdatedAt);
+  }
+
+  @Test
+  void testGroupMetaGetGroupUpdatedAt() {
+    insertMetalake(1L, "metalake1");
+    insertGroup(32L, "group32", 1L);
+
+    groupMetaMapper.touchGroupUpdatedAt(32L);
+    long expected = queryUpdatedAt("group_meta", "group_id", 32L);
+
+    GroupUpdatedAt info = groupMetaMapper.getGroupUpdatedAt("metalake1", "group32");
+    Assertions.assertNotNull(info);
+    Assertions.assertEquals(32L, info.getGroupId());
+    Assertions.assertEquals(expected, info.getUpdatedAt());
+  }
+
   @Test
   void testOwnerMetaSelectOwnerByMetadataObjectIdAndType() {
     insertMetalake(1L, "metalake1");

core/src/test/java/org/apache/gravitino/storage/relational/service/TestGroupMetaService.java

@@ -50,6 +50,7 @@
 import org.apache.gravitino.storage.relational.mapper.GroupMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
 import org.apache.gravitino.storage.relational.po.RolePO;
+import org.apache.gravitino.storage.relational.po.auth.GroupUpdatedAt;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.gravitino.storage.relational.utils.SessionUtils;
 import org.apache.gravitino.utils.NameIdentifierUtil;
@@ -552,6 +553,7 @@ void testUpdateGroup() throws IOException {
             Lists.newArrayList(role1.name(), role2.name()),
             Lists.newArrayList(role1.id(), role2.id()));
     groupMetaService.insertGroup(group1, false);
+    Assertions.assertEquals(0L, getGroupUpdatedAt(group1.name()).getUpdatedAt());
     GroupEntity actualGroup = groupMetaService.getGroupByIdentifier(group1.nameIdentifier());
     Assertions.assertEquals(group1.name(), actualGroup.name());
     Assertions.assertEquals(
@@ -592,7 +594,13 @@ void testUpdateGroup() throws IOException {
               .build();
         };
 
+    long beforeGrant = System.currentTimeMillis();
     Assertions.assertNotNull(groupMetaService.updateGroup(group1.nameIdentifier(), grantUpdater));
+    long afterGrant = System.currentTimeMillis();
+    long grantUpdatedAt = getGroupUpdatedAt(group1.name()).getUpdatedAt();
+    Assertions.assertTrue(
+        grantUpdatedAt >= beforeGrant - 1000L && grantUpdatedAt <= afterGrant + 1000L,
+        "expected DB-time group updated_at within 1s of JVM clock, got " + grantUpdatedAt);
     GroupEntity grantGroup =
         GroupMetaService.getInstance().getGroupByIdentifier(group1.nameIdentifier());
     Assertions.assertEquals(group1.id(), grantGroup.id());
@@ -630,7 +638,13 @@ void testUpdateGroup() throws IOException {
               .build();
         };
 
+    long beforeRevoke = System.currentTimeMillis();
     Assertions.assertNotNull(groupMetaService.updateGroup(group1.nameIdentifier(), revokeUpdater));
+    long afterRevoke = System.currentTimeMillis();
+    long revokeUpdatedAt = getGroupUpdatedAt(group1.name()).getUpdatedAt();
+    Assertions.assertTrue(
+        revokeUpdatedAt >= beforeRevoke - 1000L && revokeUpdatedAt <= afterRevoke + 1000L,
+        "expected DB-time group updated_at within 1s of JVM clock, got " + revokeUpdatedAt);
     GroupEntity revokeGroup =
         GroupMetaService.getInstance().getGroupByIdentifier(group1.nameIdentifier());
     Assertions.assertEquals(group1.id(), revokeGroup.id());
@@ -714,7 +728,9 @@ void testUpdateGroup() throws IOException {
               .withAuditInfo(updateAuditInfo)
               .build();
         };
+    long beforeNoUpdate = getGroupUpdatedAt(group1.name()).getUpdatedAt();
     Assertions.assertNotNull(groupMetaService.updateGroup(group1.nameIdentifier(), noUpdater));
+    Assertions.assertEquals(beforeNoUpdate, getGroupUpdatedAt(group1.name()).getUpdatedAt());
     GroupEntity noUpdaterGroup =
         GroupMetaService.getInstance().getGroupByIdentifier(group1.nameIdentifier());
     Assertions.assertEquals(group1.id(), noUpdaterGroup.id());
@@ -1042,6 +1058,11 @@ private Integer countGroupRoleRels() {
     return count;
   }
 
+  private GroupUpdatedAt getGroupUpdatedAt(String groupName) {
+    return SessionUtils.doWithCommitAndFetchResult(
+        GroupMetaMapper.class, mapper -> mapper.getGroupUpdatedAt(metalakeName, groupName));
+  }
+
   private GroupEntity createGroupEntity(
       Long id, Namespace namespace, String name, AuditInfo auditInfo) {
     return GroupEntity.builder()

scripts/h2/schema-1.3.0-h2.sql

@@ -231,8 +231,10 @@ CREATE TABLE IF NOT EXISTS `group_meta` (
     `current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'group current version',
     `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'group last version',
     `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'group deleted at',
+    `updated_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'updated at',
     PRIMARY KEY (`group_id`),
-    CONSTRAINT `uk_mid_gr_del` UNIQUE (`metalake_id`, `group_name`, `deleted_at`)
+    CONSTRAINT `uk_mid_gr_del` UNIQUE (`metalake_id`, `group_name`, `deleted_at`),
+    KEY `idx_group_meta_name_del_upd` (`metalake_id`, `group_name`, `deleted_at`, `updated_at`)
 ) ENGINE=InnoDB;
 
 CREATE TABLE IF NOT EXISTS `group_role_rel` (

scripts/h2/upgrade-1.2.0-to-1.3.0-h2.sql

@@ -20,9 +20,11 @@
 ALTER TABLE `user_meta`  ADD COLUMN `updated_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'updated at';
 ALTER TABLE `role_meta` ADD COLUMN `updated_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'updated at';
 ALTER TABLE `owner_meta` ADD COLUMN `updated_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'updated at';
+ALTER TABLE `group_meta` ADD COLUMN `updated_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'updated at';
 
 CREATE INDEX IF NOT EXISTS `idx_user_meta_name_del_upd` ON `user_meta`(`metalake_id`, `user_name`, `deleted_at`, `updated_at`);
 CREATE INDEX IF NOT EXISTS `idx_owner_meta_del_upd_obj` ON `owner_meta`(`deleted_at`, `updated_at`, `metadata_object_id`);
+CREATE INDEX IF NOT EXISTS `idx_group_meta_name_del_upd` ON `group_meta`(`metalake_id`, `group_name`, `deleted_at`, `updated_at`);
 
 CREATE TABLE IF NOT EXISTS `entity_change_log` (
   `id`            BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment id',