[Improvement](profile) add catalog info in profile

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

@@ -44,6 +44,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";
@@ -107,7 +108,7 @@ public class SummaryProfile {
     // a column, so that should not
     // add many columns here. Add to ExecutionSummary list.
     public static final ImmutableList<String> SUMMARY_CAPTIONS = 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> SUMMARY_KEYS = new ImmutableList.Builder<String>()
             .addAll(SUMMARY_CAPTIONS)
             .add(PHYSICAL_PLAN)
@@ -593,6 +594,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

@@ -42,6 +42,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;
@@ -400,6 +401,7 @@ private Map<String, String> getSummaryInfo(boolean isFinished) {
         }
         builder.taskState(isFinished ? "FINISHED" : "RUNNING");
         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

@@ -389,6 +389,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

@@ -51,5 +51,34 @@ suite('test_profile') {
     def notExistingProfileString = getProfile("-100")
     logger.info("notExistingProfileString:{}", notExistingProfileString)
     def json2 = new JsonSlurper().parseText(notExistingProfileString)
-    assertEquals("ID -100 does not exist", json2.data)
+    assertEquals("Profile -100 not found", json2.data)
+
+    sql """
+        CREATE TABLE if not exists `test_profile` (
+          `id` INT,
+          `name` varchar(32)
+        )ENGINE=OLAP
+        UNIQUE KEY(`id`)
+        DISTRIBUTED BY HASH(`id`) BUCKETS 10
+        PROPERTIES (
+            "replication_allocation" = "tag.location.default: 1"
+        );
+    """
+
+    sql "set enable_profile=true"
+    def simpleSql = "select count(*) from test_profile"
+    sql "${simpleSql}"
+    def isRecorded = false
+    def wholeString = getProfileList()
+    List profileData = new JsonSlurper().parseText(wholeString).data.rows
+    for (final def profileItem in profileData) {
+        if (profileItem["Sql Statement"].toString() == simpleSql) {
+            isRecorded = true
+            assertEquals("internal", profileItem["Default Catalog"].toString())
+        }
+    }
+    assertTrue(isRecorded)
+
+    sql """ SET enable_profile = false """
+    sql """ DROP TABLE IF EXISTS test_profile """
 }