[Enhancement] Delete shards meta from starmgr after recycling partition for cloud-native cluster (backport #56691) (#57282)

Co-authored-by: Drake Wang <[email protected]>

Author: mergify[bot]

Diff for: fe/fe-core/src/main/java/com/starrocks/lake/LakeTableHelper.java

@@ -43,8 +43,12 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 
 public class LakeTableHelper {
     private static final Logger LOG = LogManager.getLogger(LakeTableHelper.class);
@@ -160,6 +164,26 @@ static boolean removePartitionDirectory(Partition partition, long warehouseId) t
         return ret;
     }
 
+    /**
+     * delete `partition`'s all shard group meta (shards meta included) from starmanager
+     */
+    static void deleteShardGroupMeta(Partition partition)  {
+        // use Set to avoid duplicate shard group id
+        StarOSAgent starOSAgent = GlobalStateMgr.getCurrentState().getStarOSAgent();
+        Collection<PhysicalPartition> subPartitions = partition.getSubPartitions();
+        Set<Long> needRemoveShardGroupIdSet = new HashSet<>();
+        for (PhysicalPartition subPartition : subPartitions) {
+            for (MaterializedIndex index : subPartition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL)) {
+                needRemoveShardGroupIdSet.add(index.getShardGroupId());
+            }
+        }
+        if (!needRemoveShardGroupIdSet.isEmpty()) {
+            starOSAgent.deleteShardGroup(new ArrayList<>(needRemoveShardGroupIdSet));
+            LOG.debug("Deleted shard group related to partition {}, group ids: {}", partition.getId(),
+                    needRemoveShardGroupIdSet);
+        }
+    }
+
     public static boolean isSharedPartitionDirectory(PhysicalPartition physicalPartition, long warehouseId)
             throws StarClientException {
         ShardInfo shardInfo = getAssociatedShardInfo(physicalPartition, warehouseId).orElse(null);

Diff for: fe/fe-core/src/main/java/com/starrocks/lake/RecycleLakeListPartitionInfo.java

@@ -40,6 +40,7 @@ public boolean delete() {
             Warehouse warehouse = manager.getBackgroundWarehouse();
             if (LakeTableHelper.removePartitionDirectory(partition, warehouse.getId())) {
                 GlobalStateMgr.getCurrentState().getLocalMetastore().onErasePartition(partition);
+                LakeTableHelper.deleteShardGroupMeta(partition);
                 return true;
             } else {
                 return false;

Diff for: fe/fe-core/src/main/java/com/starrocks/lake/RecycleLakeRangePartitionInfo.java

@@ -43,6 +43,7 @@ public boolean delete() {
             Warehouse warehouse = manager.getBackgroundWarehouse();
             if (LakeTableHelper.removePartitionDirectory(partition, warehouse.getId())) {
                 GlobalStateMgr.getCurrentState().getLocalMetastore().onErasePartition(partition);
+                LakeTableHelper.deleteShardGroupMeta(partition);
                 return true;
             } else {
                 return false;

Diff for: fe/fe-core/src/main/java/com/starrocks/lake/RecycleLakeUnPartitionInfo.java

@@ -42,6 +42,7 @@ public boolean delete() {
             Warehouse warehouse = manager.getBackgroundWarehouse();
             if (LakeTableHelper.removePartitionDirectory(partition, warehouse.getId())) {
                 GlobalStateMgr.getCurrentState().getLocalMetastore().onErasePartition(partition);
+                LakeTableHelper.deleteShardGroupMeta(partition);
                 return true;
             } else {
                 return false;

Diff for: fe/fe-core/src/test/java/com/starrocks/lake/LakeTableHelperTest.java

@@ -14,7 +14,16 @@
 
 package com.starrocks.lake;
 
+import com.google.common.collect.Lists;
+import com.staros.proto.ShardGroupInfo;
 import com.starrocks.catalog.Database;
+import com.starrocks.catalog.DistributionInfo;
+import com.starrocks.catalog.HashDistributionInfo;
+import com.starrocks.catalog.MaterializedIndex;
+import com.starrocks.catalog.Partition;
+import com.starrocks.catalog.PartitionInfo;
+import com.starrocks.catalog.PhysicalPartition;
+import com.starrocks.catalog.SinglePartitionInfo;
 import com.starrocks.catalog.Table;
 import com.starrocks.common.Config;
 import com.starrocks.qe.ConnectContext;
@@ -24,11 +33,20 @@
 import com.starrocks.sql.ast.CreateTableStmt;
 import com.starrocks.transaction.TransactionState;
 import com.starrocks.utframe.UtFrameUtils;
+import mockit.Mock;
+import mockit.MockUp;
+import mockit.Mocked;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
 public class LakeTableHelperTest {
     private static ConnectContext connectContext;
     private static final String DB_NAME = "test_lake_table_helper";
@@ -74,4 +92,54 @@ public void testSupportCombinedTxnLog() throws Exception {
         Config.lake_use_combined_txn_log = false;
         Assert.assertFalse(LakeTableHelper.supportCombinedTxnLog(TransactionState.LoadJobSourceType.BACKEND_STREAMING));
     }
+
+    @Test
+    public void testDeleteShardGroupMeta(@Mocked StarOSAgent starOSAgent) {
+
+        new MockUp<GlobalStateMgr>() {
+            @Mock
+            public StarOSAgent getStarOSAgent() {
+                return starOSAgent;
+            }
+        };
+
+        long tableId = 1001L;
+        long partitionId = 1000L;
+        long physicalPartitionId = 1002L;
+        long groupIdToClear = 5100L;
+
+        DistributionInfo distributionInfo = new HashDistributionInfo(10, Lists.newArrayList());
+        PartitionInfo partitionInfo = new SinglePartitionInfo();
+        partitionInfo.setReplicationNum(1000L, (short) 3);
+        Partition partition =
+                new Partition(partitionId, physicalPartitionId, "p1", new MaterializedIndex(), distributionInfo);
+        Collection<PhysicalPartition> subPartitions = partition.getSubPartitions();
+        subPartitions.forEach(physicalPartition -> {
+            MaterializedIndex materializedIndex =
+                    physicalPartition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL).get(0);
+            materializedIndex.setShardGroupId(groupIdToClear);
+        });
+
+        // build shardGroupInfos
+        List<Long> allShardIds = Stream.of(1000L, 1001L, 1002L, 1003L).collect(Collectors.toList());
+        List<ShardGroupInfo> shardGroupInfos = new ArrayList<>();
+        ShardGroupInfo info = ShardGroupInfo.newBuilder()
+                .setGroupId(groupIdToClear)
+                .putLabels("tableId", String.valueOf(tableId))
+                .putProperties("createTime", String.valueOf(System.currentTimeMillis() - 86400 * 1000))
+                .addAllShardIds(allShardIds)
+                .build();
+        shardGroupInfos.add(info);
+        new MockUp<StarOSAgent>() {
+            @Mock
+            public void deleteShardGroup(List<Long> groupIds) {
+                for (long groupId : groupIds) {
+                    shardGroupInfos.removeIf(item -> item.getGroupId() == groupId);
+                }
+            }
+        };
+
+        LakeTableHelper.deleteShardGroupMeta(partition);
+        Assert.assertEquals(0, shardGroupInfos.size());
+    }
 }