[feature](catalog) support medium allocation mode

fe/fe-core/src/main/java/org/apache/doris/backup/RestoreJob.java

@@ -34,6 +34,7 @@
 import org.apache.doris.catalog.MaterializedIndex;
 import org.apache.doris.catalog.MaterializedIndex.IndexExtState;
 import org.apache.doris.catalog.MaterializedIndexMeta;
+import org.apache.doris.catalog.MediumAllocationMode;
 import org.apache.doris.catalog.OdbcCatalogResource;
 import org.apache.doris.catalog.OdbcTable;
 import org.apache.doris.catalog.OlapTable;
@@ -1539,7 +1540,8 @@ protected Partition resetTabletForRestore(OlapTable localTbl, OlapTable remoteTb
                 // replicas
                 try {
                     Pair<Map<Tag, List<Long>>, TStorageMedium> beIdsAndMedium = Env.getCurrentSystemInfo()
-                            .selectBackendIdsForReplicaCreation(replicaAlloc, nextIndexes, null, false, false);
+                            .selectBackendIdsForReplicaCreation(replicaAlloc, nextIndexes, null,
+                                    MediumAllocationMode.ADAPTIVE, false);
                     Map<Tag, List<Long>> beIds = beIdsAndMedium.first;
                     for (Map.Entry<Tag, List<Long>> entry : beIds.entrySet()) {
                         for (Long beId : entry.getValue()) {

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

@@ -43,7 +43,17 @@ public class DataProperty implements GsonPostProcessable {
     private String storagePolicy;
     @SerializedName(value = "isMutable")
     private boolean isMutable = true;
-    private boolean storageMediumSpecified;
+    // Whether the medium was explicitly requested by the user at CREATE TABLE time.
+    // ADAPTIVE: default / auto - placement may pick any available medium
+    // STRICT  : user explicitly asked for a specific medium (e.g. storage_medium=ssd)
+    //
+    // NOTE: on master this signal lived in a transient `storageMediumSpecified` boolean
+    // that was never persisted (missing @SerializedName), so it silently reverted to false
+    // after an FE restart. By persisting it here we both (a) carry the signal across
+    // restarts correctly and (b) get a typed home for the upcoming restore-side medium
+    // decision logic. See MediumAllocationMode for semantics.
+    @SerializedName(value = "mediumAllocationMode")
+    private MediumAllocationMode mediumAllocationMode = MediumAllocationMode.ADAPTIVE;
 
     private DataProperty() {
         // for persist
@@ -53,13 +63,15 @@ public DataProperty(TStorageMedium medium) {
         this.storageMedium = medium;
         this.cooldownTimeMs = MAX_COOLDOWN_TIME_MS;
         this.storagePolicy = "";
+        this.mediumAllocationMode = MediumAllocationMode.ADAPTIVE;
     }
 
     public DataProperty(DataProperty other) {
         this.storageMedium = other.storageMedium;
         this.cooldownTimeMs = other.cooldownTimeMs;
         this.storagePolicy = other.storagePolicy;
         this.isMutable = other.isMutable;
+        this.mediumAllocationMode = other.mediumAllocationMode;
     }
 
     /**
@@ -78,6 +90,7 @@ public DataProperty(TStorageMedium medium, long cooldown, String storagePolicy,
         this.cooldownTimeMs = cooldown;
         this.storagePolicy = storagePolicy;
         this.isMutable = isMutable;
+        this.mediumAllocationMode = MediumAllocationMode.ADAPTIVE;
     }
 
     public TStorageMedium getStorageMedium() {
@@ -96,8 +109,35 @@ public void setStoragePolicy(String storagePolicy) {
         this.storagePolicy = storagePolicy;
     }
 
+    public MediumAllocationMode getMediumAllocationMode() {
+        return mediumAllocationMode;
+    }
+
+    public void setMediumAllocationMode(MediumAllocationMode mode) {
+        this.mediumAllocationMode = (mode == null ? MediumAllocationMode.ADAPTIVE : mode);
+    }
+
+    /**
+     * Legacy alias kept so that callers migrating in follow-up commits do not all
+     * need to change at once. Prefer {@link #getMediumAllocationMode()} in new code.
+     *
+     * @deprecated use {@link #getMediumAllocationMode()} and compare with
+     *             {@link MediumAllocationMode#STRICT}.
+     */
+    @Deprecated
     public boolean isStorageMediumSpecified() {
-        return storageMediumSpecified;
+        return mediumAllocationMode == MediumAllocationMode.STRICT;
+    }
+
+    /**
+     * Legacy setter kept so that callers migrating in follow-up commits do not all
+     * need to change at once. Prefer {@link #setMediumAllocationMode(MediumAllocationMode)}.
+     *
+     * @deprecated use {@link #setMediumAllocationMode(MediumAllocationMode)}.
+     */
+    @Deprecated
+    public void setStorageMediumSpecified(boolean isSpecified) {
+        this.mediumAllocationMode = isSpecified ? MediumAllocationMode.STRICT : MediumAllocationMode.ADAPTIVE;
     }
 
     public boolean isMutable() {
@@ -108,17 +148,13 @@ public void setMutable(boolean mutable) {
         isMutable = mutable;
     }
 
-    public void setStorageMediumSpecified(boolean isSpecified) {
-        storageMediumSpecified = isSpecified;
-    }
-
     public void setStorageMedium(TStorageMedium medium) {
         this.storageMedium = medium;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(storageMedium, cooldownTimeMs, storagePolicy);
+        return Objects.hash(storageMedium, cooldownTimeMs, storagePolicy, mediumAllocationMode);
     }
 
     @Override
@@ -136,7 +172,8 @@ public boolean equals(Object obj) {
         return this.storageMedium == other.storageMedium
                 && this.cooldownTimeMs == other.cooldownTimeMs
                 && Strings.nullToEmpty(this.storagePolicy).equals(Strings.nullToEmpty(other.storagePolicy))
-                && this.isMutable == other.isMutable;
+                && this.isMutable == other.isMutable
+                && this.mediumAllocationMode == other.mediumAllocationMode;
     }
 
     @Override
@@ -145,13 +182,17 @@ public String toString() {
         sb.append("Storage medium[").append(this.storageMedium).append("]. ");
         sb.append("cool down[").append(TimeUtils.longToTimeString(cooldownTimeMs)).append("]. ");
         sb.append("remote storage policy[").append(this.storagePolicy).append("]. ");
+        sb.append("medium allocation mode[").append(this.mediumAllocationMode).append("]. ");
         return sb.toString();
     }
 
     @Override
     public void gsonPostProcess() throws IOException {
         // storagePolicy is a newly added field, it may be null when replaying from old version.
         this.storagePolicy = Strings.nullToEmpty(this.storagePolicy);
+        // mediumAllocationMode is a newly added field; old images won't contain it, so default to ADAPTIVE.
+        if (this.mediumAllocationMode == null) {
+            this.mediumAllocationMode = MediumAllocationMode.ADAPTIVE;
+        }
     }
-
 }

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

@@ -0,0 +1,83 @@
+// 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.doris.catalog;
+
+import org.apache.doris.common.AnalysisException;
+
+import com.google.common.base.Strings;
+
+/**
+ * Defines how Doris decides the storage medium of a data property
+ * (partition-level / table-level).
+ *
+ * <p>Hard-binding semantics with CREATE TABLE:
+ * <ul>
+ *   <li>{@code PROPERTIES("storage_medium"="ssd|hdd")} -> {@link #STRICT}</li>
+ *   <li>no {@code storage_medium} property -> {@link #ADAPTIVE}</li>
+ * </ul>
+ *
+ * <p>The mode drives replica placement:
+ * <ul>
+ *   <li>{@link #STRICT}: user explicitly requested a medium; placement must
+ *       honour it and fail if the requested medium cannot be satisfied.</li>
+ *   <li>{@link #ADAPTIVE}: medium is a hint; placement may pick any available
+ *       medium according to cluster capacity.</li>
+ * </ul>
+ */
+public enum MediumAllocationMode {
+    STRICT("strict"),
+    ADAPTIVE("adaptive");
+
+    private final String value;
+
+    MediumAllocationMode(String value) {
+        this.value = value;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public boolean isStrict() {
+        return this == STRICT;
+    }
+
+    public boolean isAdaptive() {
+        return this == ADAPTIVE;
+    }
+
+    /**
+     * Parse a user-provided string (case-insensitive, trimmed) into the enum.
+     *
+     * @throws AnalysisException if the value is null, blank or unrecognised.
+     */
+    public static MediumAllocationMode fromString(String value) throws AnalysisException {
+        String trimmed = Strings.nullToEmpty(value).trim();
+        if (trimmed.isEmpty()) {
+            throw new AnalysisException("medium_allocation_mode cannot be null or empty");
+        }
+        for (MediumAllocationMode mode : values()) {
+            if (mode.value.equalsIgnoreCase(trimmed)) {
+                return mode;
+            }
+        }
+        throw new AnalysisException(String.format(
+                "Invalid medium_allocation_mode value: '%s'. Valid options are: 'strict', 'adaptive'",
+                value));
+    }
+}

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

@@ -970,7 +970,7 @@ && getTableProperty().getDynamicPartitionProperty().getBuckets()
                             Pair<Map<Tag, List<Long>>, TStorageMedium> tag2beIdsAndMedium =
                                     Env.getCurrentSystemInfo().selectBackendIdsForReplicaCreation(
                                             replicaAlloc, nextIndexes, null,
-                                            false, false);
+                                            MediumAllocationMode.ADAPTIVE, false);
                             tag2beIds = tag2beIdsAndMedium.first;
                         }
                         for (Map.Entry<Tag, List<Long>> entry3 : tag2beIds.entrySet()) {

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

@@ -73,6 +73,17 @@ public class TableProperty implements GsonPostProcessable {
 
     private TStorageMedium storageMedium = null;
 
+    // Table-level cache of the medium allocation mode derived from the table's
+    // raw PROPERTIES map. Like {@link #storageMedium} this is rebuilt from
+    // {@code properties} on load rather than persisted as its own field, so
+    // {@link DataProperty#mediumAllocationMode} (the partition-level source of
+    // truth) remains the only persisted copy.
+    //
+    // Hard-binding at CREATE TABLE time: if the user specifies
+    // {@code PROPERTIES("storage_medium"=...)} the mode is STRICT, otherwise
+    // ADAPTIVE. See {@link MediumAllocationMode}.
+    private MediumAllocationMode mediumAllocationMode = MediumAllocationMode.ADAPTIVE;
+
     // which columns stored in RowStore column
     private List<String> rowStoreColumns;
 
@@ -163,6 +174,7 @@ public TableProperty buildProperty(short opCode) {
                 buildInMemory();
                 buildMinLoadReplicaNum();
                 buildStorageMedium();
+                buildMediumAllocationMode();
                 buildStoragePolicy();
                 buildIsBeingSynced();
                 buildCompactionPolicy();
@@ -535,6 +547,24 @@ public TStorageMedium getStorageMedium() {
         return storageMedium;
     }
 
+    /**
+     * Derive the table-level {@link MediumAllocationMode} from the raw properties
+     * map. Mirrors the hard-binding used by {@link DataProperty}: the user
+     * explicitly asked for a medium iff {@link PropertyAnalyzer#PROPERTIES_STORAGE_MEDIUM}
+     * is present and non-empty.
+     */
+    public TableProperty buildMediumAllocationMode() {
+        String storageMediumStr = properties.get(PropertyAnalyzer.PROPERTIES_STORAGE_MEDIUM);
+        mediumAllocationMode = Strings.isNullOrEmpty(storageMediumStr)
+                ? MediumAllocationMode.ADAPTIVE
+                : MediumAllocationMode.STRICT;
+        return this;
+    }
+
+    public MediumAllocationMode getMediumAllocationMode() {
+        return mediumAllocationMode;
+    }
+
     public TableProperty buildStoragePolicy() {
         storagePolicy = properties.getOrDefault(PropertyAnalyzer.PROPERTIES_STORAGE_POLICY, "");
         return this;
@@ -910,6 +940,7 @@ public void gsonPostProcess() throws IOException {
         buildInMemory();
         buildMinLoadReplicaNum();
         buildStorageMedium();
+        buildMediumAllocationMode();
         buildStorageFormat();
         buildInvertedIndexFileStorageFormat();
         buildDataSortInfo();

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

@@ -33,6 +33,7 @@
 import org.apache.doris.catalog.MaterializedIndex.IndexExtState;
 import org.apache.doris.catalog.MaterializedIndex.IndexState;
 import org.apache.doris.catalog.MaterializedIndexMeta;
+import org.apache.doris.catalog.MediumAllocationMode;
 import org.apache.doris.catalog.MetaIdGenerator.IdGeneratorBuffer;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.Partition;
@@ -108,7 +109,7 @@ protected Partition createPartitionWithIndices(long dbId, OlapTable tbl, long pa
                                                    String storagePolicy,
                                                    IdGeneratorBuffer idGeneratorBuffer,
                                                    BinlogConfig binlogConfig,
-                                                   boolean isStorageMediumSpecified)
+                                                   MediumAllocationMode mediumAllocationMode)
             throws DdlException {
         // create base index first.
         Preconditions.checkArgument(tbl.getBaseIndexId() != -1);

fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java

@@ -18,6 +18,7 @@
 package org.apache.doris.cloud.system;
 
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MediumAllocationMode;
 import org.apache.doris.catalog.ReplicaAllocation;
 import org.apache.doris.cloud.catalog.CloudEnv;
 import org.apache.doris.cloud.catalog.ComputeGroup;
@@ -220,7 +221,7 @@ public void renameVirtualComputeGroup(String computeGroupId, String oldComputeGr
     @Override
     public Pair<Map<Tag, List<Long>>, TStorageMedium> selectBackendIdsForReplicaCreation(
             ReplicaAllocation replicaAlloc, Map<Tag, Integer> nextIndexs,
-            TStorageMedium storageMedium, boolean isStorageMediumSpecified,
+            TStorageMedium storageMedium, MediumAllocationMode mediumAllocationMode,
             boolean isOnlyForCheck)
             throws DdlException {
         return Pair.of(Maps.newHashMap(), storageMedium);

fe/fe-core/src/main/java/org/apache/doris/common/util/DynamicPartitionUtil.java

@@ -26,6 +26,7 @@
 import org.apache.doris.catalog.DistributionInfo;
 import org.apache.doris.catalog.DynamicPartitionProperty;
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MediumAllocationMode;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.PartitionInfo;
 import org.apache.doris.catalog.PartitionType;
@@ -237,7 +238,7 @@ private static void checkReplicationNum(String val, Database db) throws DdlExcep
         }
         ReplicaAllocation replicaAlloc = new ReplicaAllocation(Short.valueOf(val));
         Env.getCurrentSystemInfo().selectBackendIdsForReplicaCreation(replicaAlloc, Maps.newHashMap(),
-                null, false, true);
+                null, MediumAllocationMode.ADAPTIVE, true);
     }
 
     private static void checkReplicaAllocation(ReplicaAllocation replicaAlloc, int hotPartitionNum)
@@ -248,14 +249,14 @@ private static void checkReplicaAllocation(ReplicaAllocation replicaAlloc, int h
 
         Map<Tag, Integer> nextIndexs = Maps.newHashMap();
         Env.getCurrentSystemInfo().selectBackendIdsForReplicaCreation(replicaAlloc, nextIndexs, null,
-                false, true);
+                MediumAllocationMode.ADAPTIVE, true);
         if (hotPartitionNum <= 0) {
             return;
         }
 
         try {
             Env.getCurrentSystemInfo().selectBackendIdsForReplicaCreation(replicaAlloc, nextIndexs,
-                    TStorageMedium.SSD, false, true);
+                    TStorageMedium.SSD, MediumAllocationMode.ADAPTIVE, true);
         } catch (DdlException e) {
             throw new DdlException("Failed to find enough backend for ssd storage medium. When setting "
                     + DynamicPartitionProperty.HOT_PARTITION_NUM + " > 0, the hot partitions will store "