diff --git a/docs/en/administration/management/FE_configuration.md b/docs/en/administration/management/FE_configuration.md
index 1e5697561260b..d1bb2988c9560 100644
--- a/docs/en/administration/management/FE_configuration.md
+++ b/docs/en/administration/management/FE_configuration.md
@@ -4549,9 +4549,9 @@ ADMIN SET FRONTEND CONFIG ("key" = "value");
 -->
 
 <!--
-##### stream_load_profile_collect_second
+##### load_profile_collect_threshold_second
 
-- Default: 10
+- Default: 0
 - Type: Long
 - Unit: Seconds
 - Is mutable: Yes
diff --git a/docs/zh/administration/management/FE_configuration.md b/docs/zh/administration/management/FE_configuration.md
index 20394c483a11a..33de6c3936ecd 100644
--- a/docs/zh/administration/management/FE_configuration.md
+++ b/docs/zh/administration/management/FE_configuration.md
@@ -4555,9 +4555,9 @@ Compaction Score 代表了一个表分区是否值得进行 Compaction 的评分
 -->
 
 <!--
-##### stream_load_profile_collect_second
+##### stream_load_profile_collect_threshold_second
 
-- 默认值：10
+- 默认值：0
 - 类型：Long
 - 单位：Seconds
 - 是否动态：是
diff --git a/fe/fe-core/src/main/java/com/starrocks/alter/AlterJobExecutor.java b/fe/fe-core/src/main/java/com/starrocks/alter/AlterJobExecutor.java
index 3409e8c29a037..82081ae6f05eb 100644
--- a/fe/fe-core/src/main/java/com/starrocks/alter/AlterJobExecutor.java
+++ b/fe/fe-core/src/main/java/com/starrocks/alter/AlterJobExecutor.java
@@ -373,6 +373,9 @@ public Void visitModifyTablePropertiesClause(ModifyTablePropertiesClause clause,
             } else if (properties.containsKey(PropertyAnalyzer.PROPERTIES_MUTABLE_BUCKET_NUM)) {
                 schemaChangeHandler.updateTableMeta(db, tableName.getTbl(), properties,
                         TTabletMetaType.MUTABLE_BUCKET_NUM);
+            } else if (properties.containsKey(PropertyAnalyzer.PROPERTIES_ENABLE_LOAD_PROFILE)) {
+                schemaChangeHandler.updateTableMeta(db, tableName.getTbl(), properties,
+                        TTabletMetaType.ENABLE_LOAD_PROFILE);
             } else if (properties.containsKey(PropertyAnalyzer.PROPERTIES_BINLOG_ENABLE) ||
                     properties.containsKey(PropertyAnalyzer.PROPERTIES_BINLOG_TTL) ||
                     properties.containsKey(PropertyAnalyzer.PROPERTIES_BINLOG_MAX_SIZE)) {
diff --git a/fe/fe-core/src/main/java/com/starrocks/alter/SchemaChangeHandler.java b/fe/fe-core/src/main/java/com/starrocks/alter/SchemaChangeHandler.java
index a76da34b75b03..887fac3b940f3 100644
--- a/fe/fe-core/src/main/java/com/starrocks/alter/SchemaChangeHandler.java
+++ b/fe/fe-core/src/main/java/com/starrocks/alter/SchemaChangeHandler.java
@@ -2100,6 +2100,11 @@ public void updateTableMeta(Database db, String tableName, Map<String, String> p
             if (mutableBucketNum == olapTable.getMutableBucketNum()) {
                 return;
             }
+        } else if (metaType == TTabletMetaType.ENABLE_LOAD_PROFILE) {
+            boolean enableLoadProfile = Boolean.parseBoolean(properties.get(PropertyAnalyzer.PROPERTIES_ENABLE_LOAD_PROFILE));
+            if (enableLoadProfile == olapTable.enableLoadProfile()) {
+                return;
+            }
         } else if (metaType == TTabletMetaType.PRIMARY_INDEX_CACHE_EXPIRE_SEC) {
             int primaryIndexCacheExpireSec = Integer.parseInt(properties.get(
                     PropertyAnalyzer.PROPERTIES_PRIMARY_INDEX_CACHE_EXPIRE_SEC));
diff --git a/fe/fe-core/src/main/java/com/starrocks/catalog/OlapTable.java b/fe/fe-core/src/main/java/com/starrocks/catalog/OlapTable.java
index 985f81adea8c9..c88ea0e181906 100644
--- a/fe/fe-core/src/main/java/com/starrocks/catalog/OlapTable.java
+++ b/fe/fe-core/src/main/java/com/starrocks/catalog/OlapTable.java
@@ -285,6 +285,8 @@ public enum OlapTableState {
     protected Map<Long, Long> doubleWritePartitions = new HashMap<>();
 
     // Both the following two flags are used by StarMgrMetaSyncer
+    private long lastCollectProfileTime = 0;
+
     // The flag is used to indicate whether the table shard group has changed.
     public AtomicBoolean isShardGroupChanged = new AtomicBoolean(false);
     // The flag is used to indicate whether the table is doing automatic bucketing.
@@ -2552,6 +2554,23 @@ public void setMutableBucketNum(long bucketNum) {
         tableProperty.buildMutableBucketNum();
     }
 
+    public Boolean enableLoadProfile() {
+        if (tableProperty != null) {
+            return tableProperty.enableLoadProfile();
+        }
+        return false;
+    }
+
+    public void setEnableLoadProfile(boolean enableLoadProfile) {
+        if (tableProperty == null) {
+            tableProperty = new TableProperty(new HashMap<>());
+        }
+        tableProperty
+                .modifyTableProperties(PropertyAnalyzer.PROPERTIES_ENABLE_LOAD_PROFILE,
+                        Boolean.valueOf(enableLoadProfile).toString());
+        tableProperty.buildEnableLoadProfile();
+    }
+
     public TWriteQuorumType writeQuorum() {
         if (tableProperty != null) {
             return tableProperty.writeQuorum();
@@ -3253,6 +3272,12 @@ public Map<String, String> getProperties() {
             properties.put(PropertyAnalyzer.PROPERTIES_MUTABLE_BUCKET_NUM, mutableBucketNum.toString());
         }
 
+        // enable load profile
+        Boolean enableLoadProfile = enableLoadProfile();
+        if (enableLoadProfile) {
+            properties.put(PropertyAnalyzer.PROPERTIES_ENABLE_LOAD_PROFILE, "true");
+        }
+
         // locations
         Multimap<String, String> locationsMap = getLocation();
         if (locationsMap != null) {
@@ -3529,4 +3554,12 @@ public void unlockCreatePartition(String partitionName) {
         }
     }
 
+    public long getLastCollectProfileTime() {
+        return lastCollectProfileTime;
+    }
+
+    public void updateLastCollectProfileTime() {
+        this.lastCollectProfileTime = System.currentTimeMillis();
+    }
+
 }
diff --git a/fe/fe-core/src/main/java/com/starrocks/catalog/TableProperty.java b/fe/fe-core/src/main/java/com/starrocks/catalog/TableProperty.java
index 67c1e3a66ac6f..f586017f25b6f 100644
--- a/fe/fe-core/src/main/java/com/starrocks/catalog/TableProperty.java
+++ b/fe/fe-core/src/main/java/com/starrocks/catalog/TableProperty.java
@@ -262,6 +262,8 @@ public static String valueList() {
     // the default mutable bucket number
     private long mutableBucketNum = 0;
 
+    private boolean enableLoadProfile = false;
+
     // 1. This table has been deleted. if hasDelete is false, the BE segment must don't have deleteConditions.
     //    If hasDelete is true, the BE segment maybe have deleteConditions because compaction.
     // 2. Before checkpoint, we relay delete job journal log to persist.
@@ -358,6 +360,9 @@ public TableProperty buildProperty(short opCode) {
             case OperationType.OP_MODIFY_MUTABLE_BUCKET_NUM:
                 buildMutableBucketNum();
                 break;
+            case OperationType.OP_MODIFY_ENABLE_LOAD_PROFILE:
+                buildEnableLoadProfile();
+                break;
             case OperationType.OP_MODIFY_BINLOG_CONFIG:
                 buildBinlogConfig();
                 break;
@@ -650,6 +655,13 @@ public TableProperty buildMutableBucketNum() {
         return this;
     }
 
+    public TableProperty buildEnableLoadProfile() {
+        if (properties.get(PropertyAnalyzer.PROPERTIES_ENABLE_LOAD_PROFILE) != null) {
+            enableLoadProfile = Boolean.parseBoolean(properties.get(PropertyAnalyzer.PROPERTIES_ENABLE_LOAD_PROFILE));
+        }
+        return this;
+    }
+
     public TableProperty buildEnablePersistentIndex() {
         enablePersistentIndex = Boolean.parseBoolean(
                 properties.getOrDefault(PropertyAnalyzer.PROPERTIES_ENABLE_PERSISTENT_INDEX, "false"));
@@ -906,6 +918,10 @@ public long getMutableBucketNum() {
         return mutableBucketNum;
     }
 
+    public boolean enableLoadProfile() {
+        return enableLoadProfile;
+    }
+
     public String getStorageVolume() {
         return storageVolume;
     }
@@ -1022,6 +1038,7 @@ public void gsonPostProcess() throws IOException {
         buildPartitionLiveNumber();
         buildReplicatedStorage();
         buildBucketSize();
+        buildEnableLoadProfile();
         buildBinlogConfig();
         buildBinlogAvailableVersion();
         buildDataCachePartitionDuration();
diff --git a/fe/fe-core/src/main/java/com/starrocks/common/Config.java b/fe/fe-core/src/main/java/com/starrocks/common/Config.java
index dcdf4fd4268e4..93e3955798627 100644
--- a/fe/fe-core/src/main/java/com/starrocks/common/Config.java
+++ b/fe/fe-core/src/main/java/com/starrocks/common/Config.java
@@ -2882,11 +2882,17 @@ public class Config extends ConfigBase {
     public static int routine_load_scheduler_interval_millisecond = 10000;
 
     /**
-     * Only when the stream load time exceeds this value,
+     * Only when the stream/routine load time exceeds this value,
      * the profile will be put into the profileManager
      */
+    @ConfField(mutable = true, aliases = {"stream_load_profile_collect_second"})
+    public static long stream_load_profile_collect_threshold_second = 0;
+
+    /**
+     * The interval of collecting load profile through table granularity
+     */
     @ConfField(mutable = true)
-    public static long stream_load_profile_collect_second = 10; //10s
+    public static long load_profile_collect_interval_second = 0;
 
     /**
      * If set to <= 0, means that no limitation.
diff --git a/fe/fe-core/src/main/java/com/starrocks/common/util/ProfileManager.java b/fe/fe-core/src/main/java/com/starrocks/common/util/ProfileManager.java
index d432a59e0d625..5f32d1bcb5ef1 100644
--- a/fe/fe-core/src/main/java/com/starrocks/common/util/ProfileManager.java
+++ b/fe/fe-core/src/main/java/com/starrocks/common/util/ProfileManager.java
@@ -78,6 +78,10 @@ public class ProfileManager implements MemoryTrackable {
     public static final String DEFAULT_DB = "Default Db";
     public static final String VARIABLES = "Variables";
     public static final String PROFILE_COLLECT_TIME = "Collect Profile Time";
+    public static final String LOAD_TYPE = "Load Type";
+
+    public static final String LOAD_TYPE_STREAM_LOAD = "STREAM_LOAD";
+    public static final String LOAD_TYPE_ROUTINE_LOAD = "ROUTINE_LOAD";
 
     public static final ArrayList<String> PROFILE_HEADERS = new ArrayList<>(
             Arrays.asList(QUERY_ID, USER, DEFAULT_DB, SQL_STATEMENT, QUERY_TYPE,
diff --git a/fe/fe-core/src/main/java/com/starrocks/common/util/PropertyAnalyzer.java b/fe/fe-core/src/main/java/com/starrocks/common/util/PropertyAnalyzer.java
index 9af7aee54eaa3..d57ad108ad8a9 100644
--- a/fe/fe-core/src/main/java/com/starrocks/common/util/PropertyAnalyzer.java
+++ b/fe/fe-core/src/main/java/com/starrocks/common/util/PropertyAnalyzer.java
@@ -171,6 +171,8 @@ public class PropertyAnalyzer {
 
     public static final String PROPERTIES_MUTABLE_BUCKET_NUM = "mutable_bucket_num";
 
+    public static final String PROPERTIES_ENABLE_LOAD_PROFILE = "enable_load_profile";
+
     public static final String PROPERTIES_PRIMARY_INDEX_CACHE_EXPIRE_SEC = "primary_index_cache_expire_sec";
 
     public static final String PROPERTIES_TABLET_TYPE = "tablet_type";
@@ -445,6 +447,14 @@ public static long analyzeMutableBucketNum(Map<String, String> properties) {
         }
     }
 
+    public static boolean analyzeEnableLoadProfile(Map<String, String> properties) {
+        boolean enableLoadProfile = false;
+        if (properties != null && properties.containsKey(PROPERTIES_ENABLE_LOAD_PROFILE)) {
+            enableLoadProfile = Boolean.parseBoolean(properties.get(PROPERTIES_ENABLE_LOAD_PROFILE));
+        }
+        return enableLoadProfile;
+    }
+
     public static int analyzeAutoRefreshPartitionsLimit(Map<String, String> properties, MaterializedView mv) {
         if (mv.getRefreshScheme().getType() == MaterializedView.RefreshType.MANUAL) {
             throw new SemanticException(
diff --git a/fe/fe-core/src/main/java/com/starrocks/journal/JournalEntity.java b/fe/fe-core/src/main/java/com/starrocks/journal/JournalEntity.java
index 7549e167c5215..b280fe57be39e 100644
--- a/fe/fe-core/src/main/java/com/starrocks/journal/JournalEntity.java
+++ b/fe/fe-core/src/main/java/com/starrocks/journal/JournalEntity.java
@@ -841,6 +841,7 @@ public void readFields(DataInput in) throws IOException {
             case OperationType.OP_MODIFY_REPLICATED_STORAGE:
             case OperationType.OP_MODIFY_BUCKET_SIZE:
             case OperationType.OP_MODIFY_MUTABLE_BUCKET_NUM:
+            case OperationType.OP_MODIFY_ENABLE_LOAD_PROFILE:
             case OperationType.OP_MODIFY_BINLOG_CONFIG:
             case OperationType.OP_MODIFY_BINLOG_AVAILABLE_VERSION:
             case OperationType.OP_MODIFY_ENABLE_PERSISTENT_INDEX:
diff --git a/fe/fe-core/src/main/java/com/starrocks/load/streamload/StreamLoadTask.java b/fe/fe-core/src/main/java/com/starrocks/load/streamload/StreamLoadTask.java
index 378fbcfe3b515..3a98a3c31b0d2 100644
--- a/fe/fe-core/src/main/java/com/starrocks/load/streamload/StreamLoadTask.java
+++ b/fe/fe-core/src/main/java/com/starrocks/load/streamload/StreamLoadTask.java
@@ -46,10 +46,8 @@
 import com.starrocks.persist.gson.GsonPostProcessable;
 import com.starrocks.persist.gson.GsonPreProcessable;
 import com.starrocks.persist.gson.GsonUtils;
-import com.starrocks.qe.ConnectContext;
 import com.starrocks.qe.DefaultCoordinator;
 import com.starrocks.qe.QeProcessorImpl;
-import com.starrocks.qe.SessionVariable;
 import com.starrocks.qe.scheduler.Coordinator;
 import com.starrocks.rpc.ThriftConnectionPool;
 import com.starrocks.rpc.ThriftRPCRequestExecutor;
@@ -840,7 +838,7 @@ private void unprotectedWaitCoordFinish() throws UserException {
                 }
 
                 if (coord.isEnableLoadProfile()) {
-                    collectProfile();
+                    collectProfile(false);
                 }
 
                 this.trackingUrl = coord.getTrackingUrl();
@@ -1019,7 +1017,7 @@ public void afterCommitted(TransactionState txnState, boolean txnOperated) throw
 
         // sync stream load collect profile, here we collect profile only when be has reported
         if (isSyncStreamLoad() && coord != null && coord.isProfileAlreadyReported()) {
-            collectProfile();
+            collectProfile(false);
         }
 
         writeLock();
@@ -1037,31 +1035,24 @@ public void afterCommitted(TransactionState txnState, boolean txnOperated) throw
         }
     }
 
-    public void collectProfile() {
-        long currentTimestamp = System.currentTimeMillis();
-        long totalTimeMs = currentTimestamp - createTimeMs;
-
-        // For the usage scenarios of flink cdc or routine load,
-        // the frequency of stream load maybe very high, resulting in many profiles,
-        // but we may only care about the long-duration stream load profile.
-        if (totalTimeMs < Config.stream_load_profile_collect_second * 1000) {
-            LOG.info(String.format("Load %s, totalTimeMs %d < Config.stream_load_profile_collect_second %d)",
-                    label, totalTimeMs, Config.stream_load_profile_collect_second));
-            return;
-        }
-
+    public RuntimeProfile buildTopLevelProfile(boolean isAborted) {
         RuntimeProfile profile = new RuntimeProfile("Load");
         RuntimeProfile summaryProfile = new RuntimeProfile("Summary");
         summaryProfile.addInfoString(ProfileManager.QUERY_ID, DebugUtil.printId(loadId));
         summaryProfile.addInfoString(ProfileManager.START_TIME,
                 TimeUtils.longToTimeString(createTimeMs));
 
-        summaryProfile.addInfoString(ProfileManager.END_TIME, TimeUtils.longToTimeString(System.currentTimeMillis()));
+        long currentTimestamp = System.currentTimeMillis();
+        long totalTimeMs = currentTimestamp - createTimeMs;
+        summaryProfile.addInfoString(ProfileManager.END_TIME, TimeUtils.longToTimeString(currentTimestamp));
         summaryProfile.addInfoString(ProfileManager.TOTAL_TIME, DebugUtil.getPrettyStringMs(totalTimeMs));
 
         summaryProfile.addInfoString(ProfileManager.QUERY_TYPE, "Load");
+        summaryProfile.addInfoString(ProfileManager.LOAD_TYPE, getStringByType());
+        summaryProfile.addInfoString(ProfileManager.QUERY_STATE, isAborted ? "Aborted" : "Finished");
         summaryProfile.addInfoString("StarRocks Version",
                 String.format("%s-%s", Version.STARROCKS_VERSION, Version.STARROCKS_COMMIT_HASH));
+        summaryProfile.addInfoString(ProfileManager.SQL_STATEMENT, getStmt());
         summaryProfile.addInfoString(ProfileManager.DEFAULT_DB, dbName);
 
         Map<String, String> loadCounters = coord.getLoadCounters();
@@ -1071,19 +1062,20 @@ public void collectProfile() {
             summaryProfile.addInfoString("NumRowsAbnormal", loadCounters.get(LoadEtlTask.DPP_ABNORMAL_ALL));
             summaryProfile.addInfoString("numRowsUnselected", loadCounters.get(LoadJob.UNSELECTED_ROWS));
         }
-        ConnectContext session = ConnectContext.get();
-        if (session != null) {
-            SessionVariable variables = session.getSessionVariable();
-            if (variables != null) {
-                summaryProfile.addInfoString("NonDefaultSessionVariables", variables.getNonDefaultVariablesJson());
-            }
-        }
 
         profile.addChild(summaryProfile);
+
+        return profile;
+    }
+
+
+    public void collectProfile(boolean isAborted) {
+        RuntimeProfile profile = buildTopLevelProfile(isAborted);
+
         if (coord.getQueryProfile() != null) {
             if (!isSyncStreamLoad()) {
                 coord.collectProfileSync();
-                profile.addChild(coord.buildQueryProfile(session == null || session.needMergeProfile()));
+                profile.addChild(coord.buildQueryProfile(true));
             } else {
                 profile.addChild(coord.getQueryProfile());
             }
@@ -1140,6 +1132,10 @@ public void afterAborted(TransactionState txnState, boolean txnOperated, String
             return;
         }
 
+        if (isSyncStreamLoad() && coord.isProfileAlreadyReported()) {
+            collectProfile(true);
+        }
+
         writeLock();
         try {
             if (isFinalState()) {
@@ -1351,6 +1347,10 @@ public boolean isRoutineLoadTask() {
         return type == Type.ROUTINE_LOAD;
     }
 
+    public String getStmt() {
+        return "";
+    }
+
     // for sync stream load
     public void setCoordinator(Coordinator coord) {
         this.coord = coord;
@@ -1359,9 +1359,9 @@ public void setCoordinator(Coordinator coord) {
     public String getStringByType() {
         switch (this.type) {
             case ROUTINE_LOAD:
-                return "ROUTINE_LOAD";
+                return ProfileManager.LOAD_TYPE_ROUTINE_LOAD;
             case STREAM_LOAD:
-                return "STREAM_LOAD";
+                return ProfileManager.LOAD_TYPE_STREAM_LOAD;
             case PARALLEL_STREAM_LOAD:
                 return "PARALLEL_STREAM_LOAD";
             default:
diff --git a/fe/fe-core/src/main/java/com/starrocks/persist/EditLog.java b/fe/fe-core/src/main/java/com/starrocks/persist/EditLog.java
index 68a2a8bb8eedc..b943e66df07ef 100644
--- a/fe/fe-core/src/main/java/com/starrocks/persist/EditLog.java
+++ b/fe/fe-core/src/main/java/com/starrocks/persist/EditLog.java
@@ -823,6 +823,7 @@ public void loadJournal(GlobalStateMgr globalStateMgr, JournalEntity journal)
                 case OperationType.OP_MODIFY_REPLICATED_STORAGE:
                 case OperationType.OP_MODIFY_BUCKET_SIZE:
                 case OperationType.OP_MODIFY_MUTABLE_BUCKET_NUM:
+                case OperationType.OP_MODIFY_ENABLE_LOAD_PROFILE:
                 case OperationType.OP_MODIFY_BINLOG_AVAILABLE_VERSION:
                 case OperationType.OP_MODIFY_BINLOG_CONFIG:
                 case OperationType.OP_MODIFY_ENABLE_PERSISTENT_INDEX:
@@ -1729,6 +1730,10 @@ public void logModifyMutableBucketNum(ModifyTablePropertyOperationLog info) {
         logEdit(OperationType.OP_MODIFY_MUTABLE_BUCKET_NUM, info);
     }
 
+    public void logModifyEnableLoadProfile(ModifyTablePropertyOperationLog info) {
+        logEdit(OperationType.OP_MODIFY_ENABLE_LOAD_PROFILE, info);
+    }
+
     public void logReplaceTempPartition(ReplacePartitionOperationLog info) {
         logEdit(OperationType.OP_REPLACE_TEMP_PARTITION, info);
     }
diff --git a/fe/fe-core/src/main/java/com/starrocks/persist/OperationType.java b/fe/fe-core/src/main/java/com/starrocks/persist/OperationType.java
index d515b5cfa4c6a..9e9d85f6880d7 100644
--- a/fe/fe-core/src/main/java/com/starrocks/persist/OperationType.java
+++ b/fe/fe-core/src/main/java/com/starrocks/persist/OperationType.java
@@ -691,6 +691,9 @@ public class OperationType {
     @IgnorableOnReplayFailed
     public static final short OP_MODIFY_MUTABLE_BUCKET_NUM = 11141;
 
+    @IgnorableOnReplayFailed
+    public static final short OP_MODIFY_ENABLE_LOAD_PROFILE = 11142;
+
     // external table analyze
 
     @IgnorableOnReplayFailed
diff --git a/fe/fe-core/src/main/java/com/starrocks/planner/StreamLoadPlanner.java b/fe/fe-core/src/main/java/com/starrocks/planner/StreamLoadPlanner.java
index 5bd4be3de6920..edea581b29ea3 100644
--- a/fe/fe-core/src/main/java/com/starrocks/planner/StreamLoadPlanner.java
+++ b/fe/fe-core/src/main/java/com/starrocks/planner/StreamLoadPlanner.java
@@ -280,9 +280,19 @@ public TExecPlanFragmentParams plan(TUniqueId loadId) throws UserException {
         queryOptions.setMem_limit(streamLoadInfo.getExecMemLimit());
         queryOptions.setLoad_mem_limit(streamLoadInfo.getLoadMemLimit());
 
-        if (connectContext.getSessionVariable().isEnableLoadProfile()) {
+        boolean enableLoadProfile = false;
+        enableLoadProfile |= destTable.enableLoadProfile();
+        enableLoadProfile |= connectContext.getSessionVariable().isEnableLoadProfile();
+        if (Config.load_profile_collect_interval_second > 0
+                && System.currentTimeMillis() - destTable.getLastCollectProfileTime()
+                        < Config.load_profile_collect_interval_second * 1000) {
+            enableLoadProfile = false;
+        }
+
+        if (enableLoadProfile) {
             queryOptions.setEnable_profile(true);
-            queryOptions.setLoad_profile_collect_second(Config.stream_load_profile_collect_second);
+            queryOptions.setLoad_profile_collect_second(Config.stream_load_profile_collect_threshold_second);
+            destTable.updateLastCollectProfileTime();
         }
 
         params.setQuery_options(queryOptions);
diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/StmtExecutor.java b/fe/fe-core/src/main/java/com/starrocks/qe/StmtExecutor.java
index b2b5834a3b71a..0ab5c390d9779 100644
--- a/fe/fe-core/src/main/java/com/starrocks/qe/StmtExecutor.java
+++ b/fe/fe-core/src/main/java/com/starrocks/qe/StmtExecutor.java
@@ -1334,7 +1334,8 @@ private void handleAnalyzeProfileStmt() throws IOException, UserException {
         ProfileManager.ProfileElement profileElement = ProfileManager.getInstance().getProfileElement(queryId);
         Preconditions.checkNotNull(profileElement, "query not exists");
         // For short circuit query, 'ProfileElement#plan' is null
-        if (profileElement.plan == null) {
+        if (profileElement.plan == null && profileElement.infoStrings.get(ProfileManager.QUERY_TYPE) != null &&
+                !profileElement.infoStrings.get(ProfileManager.QUERY_TYPE).equals("Load")) {
             throw new UserException(
                     "short circuit point query doesn't suppot analyze profile stmt, " +
                             "you can set it off by using  set enable_short_circuit=false");
diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/scheduler/QueryRuntimeProfile.java b/fe/fe-core/src/main/java/com/starrocks/qe/scheduler/QueryRuntimeProfile.java
index c4a12b6facb9c..8ce4a33c943ce 100644
--- a/fe/fe-core/src/main/java/com/starrocks/qe/scheduler/QueryRuntimeProfile.java
+++ b/fe/fe-core/src/main/java/com/starrocks/qe/scheduler/QueryRuntimeProfile.java
@@ -91,7 +91,7 @@ public class QueryRuntimeProfile {
     /**
      * True indicates that the profile has been reported.
      * <p> When {@link SessionVariable#isEnableLoadProfile()} is enabled,
-     * if the time costs of stream load is less than {@link Config#stream_load_profile_collect_second},
+     * if the time costs of stream load is less than {@link Config#stream_load_profile_collect_threshold_second},
      * the profile will not be reported to FE to reduce the overhead of profile under high-frequency import
      */
     private boolean profileAlreadyReported = false;
diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/scheduler/dag/FragmentInstanceExecState.java b/fe/fe-core/src/main/java/com/starrocks/qe/scheduler/dag/FragmentInstanceExecState.java
index 05b9fe7cd16e6..0f6ecb6350d04 100644
--- a/fe/fe-core/src/main/java/com/starrocks/qe/scheduler/dag/FragmentInstanceExecState.java
+++ b/fe/fe-core/src/main/java/com/starrocks/qe/scheduler/dag/FragmentInstanceExecState.java
@@ -141,6 +141,10 @@ private FragmentInstanceExecState(JobSpec jobSpec,
                                       TNetworkAddress address,
                                       long lastMissingHeartbeatTime) {
         this.jobSpec = jobSpec;
+        // fake fragment instance exec state
+        if (jobSpec == null) {
+            state = State.EXECUTING;
+        }
         this.fragmentId = fragmentId;
         this.fragmentIndex = fragmentIndex;
         this.instanceId = instanceId;
diff --git a/fe/fe-core/src/main/java/com/starrocks/server/LocalMetastore.java b/fe/fe-core/src/main/java/com/starrocks/server/LocalMetastore.java
index f8ab87e29a6f4..76cc1a71b51e9 100644
--- a/fe/fe-core/src/main/java/com/starrocks/server/LocalMetastore.java
+++ b/fe/fe-core/src/main/java/com/starrocks/server/LocalMetastore.java
@@ -4231,6 +4231,23 @@ public void modifyTableMutableBucketNum(Database db, OlapTable table, Map<String
         GlobalStateMgr.getCurrentState().getEditLog().logModifyMutableBucketNum(info);
     }
 
+    public void modifyTableEnableLoadProfile(Database db, OlapTable table, Map<String, String> properties) {
+        Locker locker = new Locker();
+        Preconditions.checkArgument(locker.isDbWriteLockHeldByCurrentThread(db));
+        TableProperty tableProperty = table.getTableProperty();
+        if (tableProperty == null) {
+            tableProperty = new TableProperty(properties);
+            table.setTableProperty(tableProperty);
+        } else {
+            tableProperty.modifyTableProperties(properties);
+        }
+        tableProperty.buildEnableLoadProfile();
+
+        ModifyTablePropertyOperationLog info =
+                new ModifyTablePropertyOperationLog(db.getId(), table.getId(), properties);
+        GlobalStateMgr.getCurrentState().getEditLog().logModifyEnableLoadProfile(info);
+    }
+
     public void modifyTablePrimaryIndexCacheExpireSec(Database db, OlapTable table, Map<String, String> properties) {
         Locker locker = new Locker();
         Preconditions.checkArgument(locker.isDbWriteLockHeldByCurrentThread(db));
@@ -4264,6 +4281,8 @@ public void modifyTableMeta(Database db, OlapTable table, Map<String, String> pr
             modifyTableMutableBucketNum(db, table, properties);
         } else if (metaType == TTabletMetaType.PRIMARY_INDEX_CACHE_EXPIRE_SEC) {
             modifyTablePrimaryIndexCacheExpireSec(db, table, properties);
+        } else if (metaType == TTabletMetaType.ENABLE_LOAD_PROFILE) {
+            modifyTableEnableLoadProfile(db, table, properties);
         }
     }
 
diff --git a/fe/fe-core/src/main/java/com/starrocks/server/OlapTableFactory.java b/fe/fe-core/src/main/java/com/starrocks/server/OlapTableFactory.java
index fe3a320ee8e64..5770e2485fca1 100644
--- a/fe/fe-core/src/main/java/com/starrocks/server/OlapTableFactory.java
+++ b/fe/fe-core/src/main/java/com/starrocks/server/OlapTableFactory.java
@@ -413,6 +413,10 @@ public Table createTable(LocalMetastore metastore, Database db, CreateTableStmt
                 throw new DdlException(e.getMessage());
             }
 
+            if (PropertyAnalyzer.analyzeBooleanProp(properties, PropertyAnalyzer.PROPERTIES_ENABLE_LOAD_PROFILE, false)) {
+                table.setEnableLoadProfile(true);
+            }
+
             // write quorum
             try {
                 table.setWriteQuorum(PropertyAnalyzer.analyzeWriteQuorum(properties));
diff --git a/fe/fe-core/src/main/java/com/starrocks/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/com/starrocks/service/FrontendServiceImpl.java
index c981b44577b03..f0e92b2f74e53 100644
--- a/fe/fe-core/src/main/java/com/starrocks/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/com/starrocks/service/FrontendServiceImpl.java
@@ -1650,20 +1650,18 @@ TExecPlanFragmentParams streamLoadPutImpl(TStreamLoadPutRequest request) throws
             StreamLoadPlanner planner = new StreamLoadPlanner(db, (OlapTable) table, streamLoadInfo);
             TExecPlanFragmentParams plan = planner.plan(streamLoadInfo.getId());
 
-            if (plan.query_options.enable_profile) {
-                StreamLoadTask streamLoadTask = GlobalStateMgr.getCurrentState().getStreamLoadMgr().
-                        getSyncSteamLoadTaskByTxnId(request.getTxnId());
-                if (streamLoadTask == null) {
-                    throw new UserException("can not find stream load task by txnId " + request.getTxnId());
-                }
+            StreamLoadTask streamLoadTask = GlobalStateMgr.getCurrentState().getStreamLoadMgr().
+                    getSyncSteamLoadTaskByTxnId(request.getTxnId());
+            if (streamLoadTask == null) {
+                throw new UserException("can not find stream load task by txnId " + request.getTxnId());
+            }
 
-                streamLoadTask.setTUniqueId(request.getLoadId());
+            streamLoadTask.setTUniqueId(request.getLoadId());
 
-                Coordinator coord = getCoordinatorFactory().createSyncStreamLoadScheduler(planner, getClientAddr());
-                streamLoadTask.setCoordinator(coord);
+            Coordinator coord = getCoordinatorFactory().createSyncStreamLoadScheduler(planner, getClientAddr());
+            streamLoadTask.setCoordinator(coord);
 
-                QeProcessorImpl.INSTANCE.registerQuery(streamLoadInfo.getId(), coord);
-            }
+            QeProcessorImpl.INSTANCE.registerQuery(streamLoadInfo.getId(), coord);
 
             plan.query_options.setLoad_job_type(TLoadJobType.STREAM_LOAD);
             // add table indexes to transaction state
diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/ExplainAnalyzer.java b/fe/fe-core/src/main/java/com/starrocks/sql/ExplainAnalyzer.java
index 7859ce419fbee..74ae30dd6ae67 100644
--- a/fe/fe-core/src/main/java/com/starrocks/sql/ExplainAnalyzer.java
+++ b/fe/fe-core/src/main/java/com/starrocks/sql/ExplainAnalyzer.java
@@ -92,6 +92,16 @@ private static int getPlanNodeId(RuntimeProfile operator) {
     }
 
     public static String analyze(ProfilingExecPlan plan, RuntimeProfile profile, List<Integer> planNodeIds) {
+        LOG.debug("plan {} profile {} planNodeIds {}", plan, profile, planNodeIds);
+        if (plan == null && profile.getChild("Summary") != null) {
+            String loadType = profile.getChild("Summary").getInfoString(ProfileManager.LOAD_TYPE);
+            if (loadType != null && (loadType.equals(ProfileManager.LOAD_TYPE_STREAM_LOAD)
+                    || loadType.equals(ProfileManager.LOAD_TYPE_ROUTINE_LOAD))) {
+                StringBuilder builder = new StringBuilder();
+                profile.prettyPrint(builder, "");
+                return builder.toString();
+            }
+        }
         ExplainAnalyzer analyzer = new ExplainAnalyzer(plan, profile, planNodeIds);
         return analyzer.analyze();
     }
diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/analyzer/AlterTableClauseAnalyzer.java b/fe/fe-core/src/main/java/com/starrocks/sql/analyzer/AlterTableClauseAnalyzer.java
index 0d5c14c46ebbb..602f0545a88da 100644
--- a/fe/fe-core/src/main/java/com/starrocks/sql/analyzer/AlterTableClauseAnalyzer.java
+++ b/fe/fe-core/src/main/java/com/starrocks/sql/analyzer/AlterTableClauseAnalyzer.java
@@ -272,6 +272,8 @@ public Void visitModifyTablePropertiesClause(ModifyTablePropertiesClause clause,
             PropertyAnalyzer.analyzeBucketSize(properties);
         } else if (properties.containsKey(PropertyAnalyzer.PROPERTIES_MUTABLE_BUCKET_NUM)) {
             PropertyAnalyzer.analyzeMutableBucketNum(properties);
+        } else if (properties.containsKey(PropertyAnalyzer.PROPERTIES_ENABLE_LOAD_PROFILE)) {
+            PropertyAnalyzer.analyzeEnableLoadProfile(properties);
         } else if (properties.containsKey(PropertyAnalyzer.PROPERTIES_BINLOG_ENABLE) ||
                 properties.containsKey(PropertyAnalyzer.PROPERTIES_BINLOG_TTL) ||
                 properties.containsKey(PropertyAnalyzer.PROPERTIES_BINLOG_MAX_SIZE)) {
diff --git a/fe/fe-core/src/test/java/com/starrocks/load/streamload/StreamLoadTaskTest.java b/fe/fe-core/src/test/java/com/starrocks/load/streamload/StreamLoadTaskTest.java
index d6cf4b3c918f1..47342a4486165 100644
--- a/fe/fe-core/src/test/java/com/starrocks/load/streamload/StreamLoadTaskTest.java
+++ b/fe/fe-core/src/test/java/com/starrocks/load/streamload/StreamLoadTaskTest.java
@@ -83,6 +83,13 @@ public void testAfterCommitted() throws UserException {
 
     @Test
     public void testAfterAborted() throws UserException {
+        streamLoadTask.setCoordinator(coord);
+        new Expectations() {
+            {
+                coord.isProfileAlreadyReported();
+                result = false;
+            }
+        };
         TransactionState txnState = new TransactionState();
         boolean txnOperated = true;
 
@@ -162,4 +169,26 @@ public void testSetLoadStateWithRLTaskTxnCommitAttachment() {
         Assert.assertEquals("http://error.log.rl", loadInfo.getUrl());
         Assert.assertEquals("Another error message", loadInfo.getError_msg());
     }
+
+    public void testBuildProfile() throws UserException {
+        streamLoadTask.setCoordinator(coord);
+        new Expectations() {
+            {
+                coord.isProfileAlreadyReported();
+                result = true;
+                coord.getQueryProfile();
+                result = null;
+            }
+        };
+        TUniqueId labelId = new TUniqueId(4, 5);
+        streamLoadTask.setTUniqueId(labelId);
+        QeProcessorImpl.INSTANCE.registerQuery(streamLoadTask.getTUniqueId(), coord);
+        Assert.assertEquals(1, QeProcessorImpl.INSTANCE.getCoordinatorCount());
+
+        TransactionState txnState = new TransactionState();
+        boolean txnOperated = true;
+        streamLoadTask.afterCommitted(txnState, txnOperated);
+        Assert.assertEquals(0, QeProcessorImpl.INSTANCE.getCoordinatorCount());
+
+    }
 }
diff --git a/fe/fe-core/src/test/java/com/starrocks/persist/OperationTypeTest.java b/fe/fe-core/src/test/java/com/starrocks/persist/OperationTypeTest.java
index ebc49593ef367..be7848e187641 100644
--- a/fe/fe-core/src/test/java/com/starrocks/persist/OperationTypeTest.java
+++ b/fe/fe-core/src/test/java/com/starrocks/persist/OperationTypeTest.java
@@ -147,6 +147,7 @@ public void testRecoverableOperations() {
         Assert.assertTrue(OperationType.IGNORABLE_OPERATIONS.contains(OperationType.OP_MODIFY_TABLE_CONSTRAINT_PROPERTY));
         Assert.assertTrue(OperationType.IGNORABLE_OPERATIONS.contains(OperationType.OP_MODIFY_BUCKET_SIZE));
         Assert.assertTrue(OperationType.IGNORABLE_OPERATIONS.contains(OperationType.OP_MODIFY_MUTABLE_BUCKET_NUM));
+        Assert.assertTrue(OperationType.IGNORABLE_OPERATIONS.contains(OperationType.OP_MODIFY_ENABLE_LOAD_PROFILE));
         Assert.assertTrue(OperationType.IGNORABLE_OPERATIONS.contains(OperationType.OP_ADD_EXTERNAL_ANALYZE_STATUS));
         Assert.assertTrue(OperationType.IGNORABLE_OPERATIONS.contains(OperationType.OP_REMOVE_EXTERNAL_ANALYZE_STATUS));
         Assert.assertTrue(OperationType.IGNORABLE_OPERATIONS.contains(OperationType.OP_ADD_EXTERNAL_ANALYZER_JOB));
diff --git a/gensrc/thrift/AgentService.thrift b/gensrc/thrift/AgentService.thrift
index 80ded29e314a4..8f3f05c46aa59 100644
--- a/gensrc/thrift/AgentService.thrift
+++ b/gensrc/thrift/AgentService.thrift
@@ -414,7 +414,8 @@ enum TTabletMetaType {
     BUCKET_SIZE,
     PRIMARY_INDEX_CACHE_EXPIRE_SEC,
     STORAGE_TYPE,
-    MUTABLE_BUCKET_NUM
+    MUTABLE_BUCKET_NUM,
+    ENABLE_LOAD_PROFILE
 }
 
 struct TTabletMetaInfo {
diff --git a/test/common/data/stream_load/kv.csv b/test/common/data/stream_load/kv.csv
new file mode 100644
index 0000000000000..01e79c32a8c99
--- /dev/null
+++ b/test/common/data/stream_load/kv.csv
@@ -0,0 +1,3 @@
+1
+2
+3
diff --git a/test/sql/test_load_profile/R/test_load_profile b/test/sql/test_load_profile/R/test_load_profile
new file mode 100644
index 0000000000000..efc3b8bc1d9e3
--- /dev/null
+++ b/test/sql/test_load_profile/R/test_load_profile
@@ -0,0 +1,86 @@
+-- name: test_table_profile
+create database db_${uuid0};
+-- result:
+-- !result
+use db_${uuid0};
+-- result:
+-- !result
+create table t(k int) properties('enable_load_profile'='true');
+-- result:
+-- !result
+show create table t;
+-- result:
+t	CREATE TABLE `t` (
+  `k` int(11) NULL COMMENT ""
+) ENGINE=OLAP 
+DUPLICATE KEY(`k`)
+DISTRIBUTED BY RANDOM
+PROPERTIES (
+"bucket_size" = "4294967296",
+"compression" = "LZ4",
+"enable_load_profile" = "true",
+"fast_schema_evolution" = "true",
+"replicated_storage" = "true",
+"replication_num" = "3"
+);
+-- !result
+alter table t set ('enable_load_profile'='false');
+-- result:
+-- !result
+show create table t;
+-- result:
+t	CREATE TABLE `t` (
+  `k` int(11) NULL COMMENT ""
+) ENGINE=OLAP 
+DUPLICATE KEY(`k`)
+DISTRIBUTED BY RANDOM
+PROPERTIES (
+"bucket_size" = "4294967296",
+"compression" = "LZ4",
+"fast_schema_evolution" = "true",
+"replicated_storage" = "true",
+"replication_num" = "3"
+);
+-- !result
+shell: curl --location-trusted -u root: -T ${root_path}/lib/../common/data/stream_load/kv.csv -XPUT ${url}/api/db_${uuid0}/t/_stream_load
+-- result:
+0
+{
+    "Status": "Success",
+    "Message": "OK"
+}
+-- !result
+show profilelist;
+-- result:
+-- !result
+alter table t set ('enable_load_profile'='true');
+-- result:
+-- !result
+show create table t;
+-- result:
+t	CREATE TABLE `t` (
+  `k` int(11) NULL COMMENT ""
+) ENGINE=OLAP 
+DUPLICATE KEY(`k`)
+DISTRIBUTED BY RANDOM
+PROPERTIES (
+"bucket_size" = "4294967296",
+"compression" = "LZ4",
+"enable_load_profile" = "true",
+"fast_schema_evolution" = "true",
+"replicated_storage" = "true",
+"replication_num" = "3"
+);
+-- !result
+shell: curl --location-trusted -u root: -T ${root_path}/lib/../common/data/stream_load/kv.csv -XPUT ${url}/api/db_${uuid0}/t/_stream_load
+-- result:
+0
+{
+    "Status": "Success",
+    "Message": "OK"
+}
+-- !result
+show profilelist;
+-- result:
+[REGEX].*Finished.*	
+-- !result
diff --git a/test/sql/test_load_profile/T/test_load_profile b/test/sql/test_load_profile/T/test_load_profile
index 6a9ddbbc7210e..3f3af58bf2fe4 100644
--- a/test/sql/test_load_profile/T/test_load_profile
+++ b/test/sql/test_load_profile/T/test_load_profile
@@ -1,3 +1,16 @@
 -- name: test_table_profile
-create table t(k int, v int);
+create database db_${uuid0};
+use db_${uuid0};
+
+create table t(k int) properties('enable_load_profile'='true');
+show create table t;
+alter table t set ('enable_load_profile'='false');
+show create table t;
+
+shell: curl --location-trusted -u root: -T ${root_path}/lib/../common/data/stream_load/kv.csv -XPUT ${url}/api/db_${uuid0}/t/_stream_load
+show profilelist;
+alter table t set ('enable_load_profile'='true');
+show create table t;
+shell: curl --location-trusted -u root: -T ${root_path}/lib/../common/data/stream_load/kv.csv -XPUT ${url}/api/db_${uuid0}/t/_stream_load
+show profilelist;