[fix](storage) check partition storage policy resource before BE update

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

@@ -94,6 +94,8 @@
 import org.apache.doris.persist.ModifyTableEngineOperationLog;
 import org.apache.doris.persist.ModifyTablePropertyOperationLog;
 import org.apache.doris.persist.ReplaceTableOperationLog;
+import org.apache.doris.policy.Policy;
+import org.apache.doris.policy.PolicyTypeEnum;
 import org.apache.doris.policy.StoragePolicy;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.ConnectContextUtil;
@@ -647,9 +649,11 @@ public void processAlterTable(AlterTableCommand command) throws UserException {
                 ModifyPartitionOp clause = ((ModifyPartitionOp) alterOp);
                 Map<String, String> properties = clause.getProperties();
                 List<String> partitionNames = clause.getPartitionNames();
+                OlapTable olapTable = (OlapTable) tableIf;
+                checkModifyPartitionStoragePolicyResource(
+                        olapTable, partitionNames, properties, clause.isTempPartition());
                 ((SchemaChangeHandler) schemaChangeHandler).updatePartitionsProperties(
                         db, tableName, partitionNames, properties);
-                OlapTable olapTable = (OlapTable) tableIf;
                 olapTable.writeLockOrDdlException();
                 try {
                     modifyPartitionsProperty(db, olapTable, partitionNames, properties, clause.isTempPartition());
@@ -916,6 +920,56 @@ private void processRename(Database db, Table table, List<AlterOp> alterOps) thr
         }
     }
 
+    private void checkModifyPartitionStoragePolicyResource(OlapTable olapTable, List<String> partitionNames,
+            Map<String, String> properties, boolean isTempPartition) throws AnalysisException {
+        if (!PropertyAnalyzer.hasStoragePolicy(properties)) {
+            return;
+        }
+
+        String newStoragePolicy = PropertyAnalyzer.analyzeStoragePolicy(properties);
+        if (Strings.isNullOrEmpty(newStoragePolicy)) {
+            return;
+        }
+
+        Optional<Policy> newPolicy = Env.getCurrentEnv().getPolicyMgr()
+                .findPolicy(newStoragePolicy, PolicyTypeEnum.STORAGE);
+        if (!newPolicy.isPresent() || !(newPolicy.get() instanceof StoragePolicy)) {
+            return;
+        }
+
+        olapTable.readLock();
+        try {
+            PartitionInfo partitionInfo = olapTable.getPartitionInfo();
+            for (String partitionName : partitionNames) {
+                Partition partition = olapTable.getPartition(partitionName, isTempPartition);
+                if (partition == null) {
+                    continue;
+                }
+
+                DataProperty dataProperty = partitionInfo.getDataProperty(partition.getId());
+                String oldStoragePolicy = dataProperty.getStoragePolicy();
+                if (Strings.isNullOrEmpty(oldStoragePolicy) || oldStoragePolicy.equals(newStoragePolicy)) {
+                    continue;
+                }
+
+                Optional<Policy> oldPolicy = Env.getCurrentEnv().getPolicyMgr()
+                        .findPolicy(oldStoragePolicy, PolicyTypeEnum.STORAGE);
+                if (!oldPolicy.isPresent() || !(oldPolicy.get() instanceof StoragePolicy)) {
+                    continue;
+                }
+
+                String newResource = ((StoragePolicy) newPolicy.get()).getStorageResource();
+                String oldResource = ((StoragePolicy) oldPolicy.get()).getStorageResource();
+                if (!newResource.equals(oldResource)) {
+                    throw new AnalysisException("currently do not support change origin "
+                            + "storage policy to another one with different resource: ");
+                }
+            }
+        } finally {
+            olapTable.readUnlock();
+        }
+    }
+
     /**
      * Batch update partitions' properties
      * caller should hold the table lock

regression-test/suites/cold_heat_separation/empty_table_use_policy/modify_partition_add_policy.groovy

@@ -112,6 +112,61 @@ suite("add_table_policy_by_modify_partition") {
         "cooldown_datetime" = "$cooldownTime"
         );
     """
+    sql """ALTER STORAGE POLICY tmp2 PROPERTIES("cooldown_datetime" = "$cooldownTime")"""
+
+    try_sql """
+        CREATE RESOURCE IF NOT EXISTS "test_modify_partition_table_use_resource_diff"
+        PROPERTIES(
+            "type"="s3",
+            "AWS_REGION" = "bj",
+            "AWS_ENDPOINT" = "bj.s3.comaaaa",
+            "AWS_ROOT_PATH" = "path/to/rootaaaa",
+            "AWS_SECRET_KEY" = "aaaa",
+            "AWS_ACCESS_KEY" = "bbba",
+            "AWS_BUCKET" = "test-bucket-diff",
+            "s3_validity_check" = "false"
+        );
+    """
+    try_sql """
+        CREATE STORAGE POLICY IF NOT EXISTS created_create_table_partition_alter_policy_diff
+        PROPERTIES(
+        "storage_resource" = "test_modify_partition_table_use_resource_diff",
+        "cooldown_datetime" = "$cooldownTime"
+        );
+    """
+    sql """ALTER STORAGE POLICY created_create_table_partition_alter_policy_diff PROPERTIES("cooldown_datetime" = "$cooldownTime")"""
+    assertEquals(storage_exist.call("created_create_table_partition_alter_policy_diff"), true)
+
+    def partitionsAfterSetPolicy = sql_return_maparray """
+    show partitions from create_table_partition
+    """
+    for (def par in partitionsAfterSetPolicy) {
+        assertTrue(par.RemoteStoragePolicy == "created_create_table_partition_alter_policy")
+    }
+
+    def alter_table_partition_diff_resource_result = try_sql """
+        ALTER TABLE create_table_partition MODIFY PARTITION (*) SET("storage_policy"="created_create_table_partition_alter_policy_diff");
+    """
+    assertEquals(alter_table_partition_diff_resource_result, null)
+
+    def partitionsAfterFailedAlter = sql_return_maparray """
+    show partitions from create_table_partition
+    """
+    for (def par in partitionsAfterFailedAlter) {
+        assertTrue(par.RemoteStoragePolicy == "created_create_table_partition_alter_policy")
+    }
+
+    def alter_table_partition_same_resource_result = try_sql """
+        ALTER TABLE create_table_partition MODIFY PARTITION (*) SET("storage_policy"="tmp2");
+    """
+    assertEquals(alter_table_partition_same_resource_result.size(), 1);
+
+    def partitionsAfterSameResourceAlter = sql_return_maparray """
+    show partitions from create_table_partition
+    """
+    for (def par in partitionsAfterSameResourceAlter) {
+        assertTrue(par.RemoteStoragePolicy == "tmp2")
+    }
 
     sql """
     CREATE TABLE create_table_partion_use_created_policy_test
@@ -148,9 +203,15 @@ suite("add_table_policy_by_modify_partition") {
     DROP STORAGE POLICY created_create_table_partition_alter_policy
     """
     sql """
+    DROP STORAGE POLICY created_create_table_partition_alter_policy_diff
+    """
+    sql """
     DROP STORAGE POLICY tmp2
     """
     sql """
     DROP RESOURCE test_modify_partition_table_use_resource
     """
+    sql """
+    DROP RESOURCE test_modify_partition_table_use_resource_diff
+    """
 }