[enhance](policy) Support to change table's storage policy if the two policy has same resource (#23665)

fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java

@@ -54,11 +54,9 @@ import org.apache.doris.catalog.OdbcTable;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.Partition;
 import org.apache.doris.catalog.PartitionInfo;
-import org.apache.doris.catalog.Replica;
 import org.apache.doris.catalog.ReplicaAllocation;
 import org.apache.doris.catalog.Table;
 import org.apache.doris.catalog.TableIf.TableType;
-import org.apache.doris.catalog.Tablet;
 import org.apache.doris.catalog.View;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
@@ -189,20 +187,21 @@ public class Alter {
         boolean needProcessOutsideTableLock = false;
         if (currentAlterOps.checkTableStoragePolicy(alterClauses)) {
             String tableStoragePolicy = olapTable.getStoragePolicy();
-            if (!tableStoragePolicy.isEmpty()) {
+            String currentStoragePolicy = currentAlterOps.getTableStoragePolicy(alterClauses);
+
+            // If the two policy has one same resource, then it's safe for the table to change policy
+            // There would only be the cooldown ttl or cooldown time would be affected
+            if (!Env.getCurrentEnv().getPolicyMgr()
+                    .checkStoragePolicyIfSameResource(tableStoragePolicy, currentStoragePolicy)
+                    && !tableStoragePolicy.isEmpty()) {
                 for (Partition partition : olapTable.getAllPartitions()) {
-                    for (Tablet tablet : partition.getBaseIndex().getTablets()) {
-                        for (Replica replica : tablet.getReplicas()) {
-                            if (replica.getRowCount() > 0 || replica.getDataSize() > 0) {
-                                throw new DdlException("Do not support alter table's storage policy , this table ["
-                                        + olapTable.getName() + "] has storage policy " + tableStoragePolicy
-                                        + ", the table need to be empty.");
-                            }
-                        }
+                    if (Partition.PARTITION_INIT_VERSION < partition.getVisibleVersion()) {
+                        throw new DdlException("Do not support alter table's storage policy , this table ["
+                                + olapTable.getName() + "] has storage policy " + tableStoragePolicy
+                                + ", the table need to be empty.");
                     }
                 }
             }
-            String currentStoragePolicy = currentAlterOps.getTableStoragePolicy(alterClauses);
             // check currentStoragePolicy resource exist.
             Env.getCurrentEnv().getPolicyMgr().checkStoragePolicyExist(currentStoragePolicy);
 

fe/fe-core/src/main/java/org/apache/doris/catalog/DataProperty.java

@@ -98,6 +98,10 @@ public class DataProperty implements Writable, GsonPostProcessable {
         return storagePolicy;
     }
 
+    public void setStoragePolicy(String storagePolicy) {
+        this.storagePolicy = storagePolicy;
+    }
+
     public boolean isStorageMediumSpecified() {
         return storageMediumSpecified;
     }

fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java

@@ -1834,6 +1834,7 @@ public class OlapTable extends Table {
         TableProperty tableProperty = getOrCreatTableProperty();
         tableProperty.modifyTableProperties(PropertyAnalyzer.PROPERTIES_STORAGE_POLICY, storagePolicy);
         tableProperty.buildStoragePolicy();
+        partitionInfo.refreshTableStoragePolicy(storagePolicy);
     }
 
     public String getStoragePolicy() {

fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionInfo.java

@@ -230,6 +230,13 @@ public class PartitionInfo implements Writable {
         idToDataProperty.put(partitionId, newDataProperty);
     }
 
+    public void refreshTableStoragePolicy(String storagePolicy) {
+        idToStoragePolicy.replaceAll((k, v) -> storagePolicy);
+        idToDataProperty.entrySet().forEach(entry -> {
+            entry.getValue().setStoragePolicy(storagePolicy);
+        });
+    }
+
     public String getStoragePolicy(long partitionId) {
         return idToStoragePolicy.getOrDefault(partitionId, "");
     }

fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java

@@ -2223,7 +2223,12 @@ public class InternalCatalog implements CatalogIf<Database> {
                     "Can not create UNIQUE KEY table that enables Merge-On-write"
                      + " with storage policy(" + storagePolicy + ")");
         }
-        olapTable.setStoragePolicy(storagePolicy);
+        // Consider one situation: if the table has no storage policy but some partitions
+        // have their own storage policy then it might be erased by the following function.
+        // So we only set the storage policy if the table's policy is not null or empty
+        if (!Strings.isNullOrEmpty(storagePolicy)) {
+            olapTable.setStoragePolicy(storagePolicy);
+        }
 
         TTabletType tabletType;
         try {

fe/fe-core/src/main/java/org/apache/doris/policy/PolicyMgr.java

@@ -533,4 +533,15 @@ public class PolicyMgr implements Writable {
             readUnlock();
         }
     }
+
+    public boolean checkStoragePolicyIfSameResource(String policyName, String anotherPolicyName) {
+        Optional<Policy> policy = findPolicy(policyName, PolicyTypeEnum.STORAGE);
+        Optional<Policy> policy1 = findPolicy(anotherPolicyName, PolicyTypeEnum.STORAGE);
+        if (policy1.isPresent() && policy.isPresent()) {
+            StoragePolicy storagePolicy = (StoragePolicy) policy.get();
+            StoragePolicy storagePolicy1 = (StoragePolicy) policy1.get();
+            return storagePolicy1.getStorageResource().equals(storagePolicy.getStorageResource());
+        }
+        return false;
+    }
 }