[Improvement](profile) add catalog info in profile (#38298)

bp #38283

Author: Yulei-Yang

fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java

@@ -41,6 +41,7 @@ public class SummaryProfile {
     public static final String TOTAL_TIME = "Total";
     public static final String TASK_STATE = "Task State";
     public static final String USER = "User";
+    public static final String DEFAULT_CATALOG = "Default Catalog";
     public static final String DEFAULT_DB = "Default Db";
     public static final String SQL_STATEMENT = "Sql Statement";
     public static final String IS_CACHED = "Is Cached";
@@ -78,7 +79,7 @@ public class SummaryProfile {
     // a column, so that should not
     // add many columns here. Add to ExcecutionSummary list.
     public static final ImmutableList<String> SUMMARY_KEYS = ImmutableList.of(PROFILE_ID, TASK_TYPE,
-            START_TIME, END_TIME, TOTAL_TIME, TASK_STATE, USER, DEFAULT_DB, SQL_STATEMENT);
+            START_TIME, END_TIME, TOTAL_TIME, TASK_STATE, USER, DEFAULT_CATALOG, DEFAULT_DB, SQL_STATEMENT);
 
     public static final ImmutableList<String> EXECUTION_SUMMARY_KEYS = ImmutableList.of(
             WORKLOAD_GROUP,
@@ -345,6 +346,11 @@ public SummaryBuilder user(String val) {
             return this;
         }
 
+        public SummaryBuilder defaultCatalog(String val) {
+            map.put(DEFAULT_CATALOG, val);
+            return this;
+        }
+
         public SummaryBuilder defaultDb(String val) {
             map.put(DEFAULT_DB, val);
             return this;

fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java

@@ -41,6 +41,7 @@
 import org.apache.doris.common.util.MetaLockUtils;
 import org.apache.doris.common.util.ProfileManager.ProfileType;
 import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.datasource.property.constants.S3Properties;
 import org.apache.doris.load.BrokerFileGroup;
 import org.apache.doris.load.BrokerFileGroupAggInfo.FileGroupAggKey;
@@ -348,6 +349,7 @@ private Map<String, String> getSummaryInfo(boolean isFinished) {
         }
         builder.taskState("FINISHED");
         builder.user(getUserInfo() != null ? getUserInfo().getQualifiedUser() : "N/A");
+        builder.defaultCatalog(InternalCatalog.INTERNAL_CATALOG_NAME);
         builder.defaultDb(getDefaultDb());
         builder.sqlStatement(getOriginStmt().originStmt);
         return builder.build();

fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java

@@ -332,6 +332,7 @@ private Map<String, String> getSummaryInfo(boolean isFinished) {
         builder.taskState(!isFinished && context.getState().getStateType().equals(MysqlStateType.OK) ? "RUNNING"
                 : context.getState().toString());
         builder.user(context.getQualifiedUser());
+        builder.defaultCatalog(context.getCurrentCatalog().getName());
         builder.defaultDb(context.getDatabase());
         builder.workloadGroup(context.getWorkloadGroupName());
         builder.sqlStatement(originStmt.originStmt);

regression-test/suites/query_profile/test_profile.groovy

@@ -138,7 +138,7 @@ suite('test_profile') {
 
         assertNotNull(stmt_query_info["Profile ID"])
         assertNotEquals(stmt_query_info["Profile ID"].toString(), "N/A".toString())
-        
+        assertNotNull(stmt_query_info["Default Catalog"])
         assertEquals(stmt_query_info['Sql Statement'].toString(), 
            """ SELECT * FROM ${table} WHERE cost ${ops[insert_order]} ${nums[insert_order]} """.toString())
     }