diff --git a/fe/fe-core/src/main/java/com/starrocks/common/AuditLog.java b/fe/fe-core/src/main/java/com/starrocks/common/AuditLog.java
index 043aa16e2e253..4a14fbcc7968e 100644
--- a/fe/fe-core/src/main/java/com/starrocks/common/AuditLog.java
+++ b/fe/fe-core/src/main/java/com/starrocks/common/AuditLog.java
@@ -25,6 +25,7 @@ public class AuditLog {
 
     public static final AuditLog SLOW_AUDIT = new AuditLog("audit.slow_query");
     public static final AuditLog QUERY_AUDIT = new AuditLog("audit.query");
+    public static final AuditLog BIG_QUERY_AUDIT = new AuditLog("big_query.query");
 
     private Logger logger;
 
@@ -36,6 +37,10 @@ public static AuditLog getSlowAudit() {
         return SLOW_AUDIT;
     }
 
+    public static AuditLog getBigQueryAudit() {
+        return BIG_QUERY_AUDIT;
+    }
+
     public AuditLog(String auditName) {
         logger = LogManager.getLogger(auditName);
     }
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 6944311b5591a..1a3ad867a7cda 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
@@ -157,6 +157,47 @@ public class Config extends ConfigBase {
     @ConfField
     public static String dump_log_delete_age = "7d";
 
+    /**
+     * big_query_log_dir:
+     * This specifies FE big query log dir.
+     * Dump log fe.big_query.log contains all information about big query.
+     * The structure of each log record is very similar to the audit log.
+     * If the cpu cost of a query exceeds big_query_log_cpu_second_threshold,
+     * or scan rows exceeds big_query_log_scan_rows_threshold,
+     * or scan bytes exceeds big_query_log_scan_bytes_threshold,
+     * we will consider it as a big query.
+     * These thresholds are defined by the user.
+     * <p>
+     * big_query_log_roll_num:
+     * Maximal FE log files to be kept within an big_query_log_roll_interval.
+     * <p>
+     * big_query_log_modules:
+     * Informations for all big queries.
+     * <p>
+     * big_query_log_roll_interval:
+     * DAY:  log suffix is yyyyMMdd
+     * HOUR: log suffix is yyyyMMddHH
+     * <p>
+     * big_query_log_delete_age:
+     * default is 7 days, if log's last modify time is 7 days ago, it will be deleted.
+     * support format:
+     * 7d      7 days
+     * 10h     10 hours
+     * 60m     60 mins
+     * 120s    120 seconds
+     */
+    @ConfField
+    public static String big_query_log_dir = StarRocksFE.STARROCKS_HOME_DIR + "/log";
+    @ConfField
+    public static int big_query_log_roll_num = 10;
+    @ConfField
+    public static String[] big_query_log_modules = {"query"};
+    @ConfField
+    public static String big_query_log_roll_interval = "DAY";
+    @ConfField
+    public static String big_query_log_delete_age = "7d";
+
+
     /**
      * plugin_dir:
      * plugin install directory
diff --git a/fe/fe-core/src/main/java/com/starrocks/common/Log4jConfig.java b/fe/fe-core/src/main/java/com/starrocks/common/Log4jConfig.java
index c9a0bacb5a78f..3f357d8fd6736 100644
--- a/fe/fe-core/src/main/java/com/starrocks/common/Log4jConfig.java
+++ b/fe/fe-core/src/main/java/com/starrocks/common/Log4jConfig.java
@@ -103,6 +103,21 @@ public class Log4jConfig extends XmlConfiguration {
             "        </Delete>\n" +
             "      </DefaultRolloverStrategy>\n" +
             "    </RollingFile>\n" +
+            "    <RollingFile name=\"BigQueryFile\" fileName=\"${big_query_log_dir}/fe.big_query.log\" filePattern=\"${big_query_log_dir}/fe.big_query.log.${big_query_file_pattern}-%i\">\n" +
+            "      <PatternLayout charset=\"UTF-8\">\n" +
+            "        <Pattern>%d{yyyy-MM-dd HH:mm:ss,SSS} [%c{1}] %m%n</Pattern>\n" +
+            "      </PatternLayout>\n" +
+            "      <Policies>\n" +
+            "        <TimeBasedTriggeringPolicy/>\n" +
+            "        <SizeBasedTriggeringPolicy size=\"${big_query_roll_maxsize}MB\"/>\n" +
+            "      </Policies>\n" +
+            "      <DefaultRolloverStrategy max=\"${sys_roll_num}\" fileIndex=\"min\">\n" +
+            "        <Delete basePath=\"${big_query_log_dir}/\" maxDepth=\"1\">\n" +
+            "          <IfFileName glob=\"fe.big_query.log.*\" />\n" +
+            "          <IfLastModified age=\"${big_query_log_delete_age}\" />\n" +
+            "        </Delete>\n" +
+            "      </DefaultRolloverStrategy>\n" +
+            "    </RollingFile>\n" +
             "  </Appenders>\n" +
             "  <Loggers>\n" +
             "    <Root level=\"${sys_log_level}\">\n" +
@@ -115,6 +130,9 @@ public class Log4jConfig extends XmlConfiguration {
             "    <Logger name=\"dump\" level=\"ERROR\" additivity=\"false\">\n" +
             "      <AppenderRef ref=\"dumpFile\"/>\n" +
             "    </Logger>\n" +
+            "    <Logger name=\"big_query\" level=\"ERROR\" additivity=\"false\">\n" +
+            "      <AppenderRef ref=\"BigQueryFile\"/>\n" +
+            "    </Logger>\n" +
             "    <Logger name=\"org.apache.thrift\" level=\"DEBUG\"> \n" +
             "      <AppenderRef ref=\"Sys\"/>\n" +
             "    </Logger>\n" +
@@ -136,6 +154,7 @@ public class Log4jConfig extends XmlConfiguration {
     private static String[] verboseModules;
     private static String[] auditModules;
     private static String[] dumpModules;
+    private static String[] bigQueryModules;
 
     private static void reconfig() throws IOException {
         String newXmlConfTemplate = xmlConfTemplate;
@@ -191,6 +210,20 @@ private static void reconfig() throws IOException {
             throw new IOException("dump_log_roll_interval config error: " + Config.dump_log_roll_interval);
         }
 
+        // big query log config
+        String bigQueryLogDir = Config.big_query_log_dir;
+        String bigQueryLogRollPattern = "%d{yyyyMMdd}";
+        String bigQueryLogRollNum = String.valueOf(Config.big_query_log_roll_num);
+        String bigQueryLogRollMaxSize = String.valueOf(Config.log_roll_size_mb);
+        String bigQueryLogDeleteAge = String.valueOf(Config.big_query_log_delete_age);
+        if (Config.big_query_log_roll_interval.equals("HOUR")) {
+            bigQueryLogRollPattern = "%d{yyyyMMddHH}";
+        } else if (Config.big_query_log_roll_interval.equals("DAY")) {
+            bigQueryLogRollPattern = "%d{yyyyMMdd}";
+        } else {
+            throw new IOException("big_query_log_roll_interval config error: " + Config.big_query_log_roll_interval);
+        }
+
         // verbose modules and audit log modules
         StringBuilder sb = new StringBuilder();
         for (String s : verboseModules) {
@@ -202,6 +235,9 @@ private static void reconfig() throws IOException {
         for (String s : dumpModules) {
             sb.append("<Logger name='dump." + s + "' level='INFO'/>");
         }
+        for (String s : bigQueryModules) {
+            sb.append("<Logger name='big_query." + s + "' level='INFO'/>");
+        }
 
         newXmlConfTemplate = newXmlConfTemplate.replaceAll("<!--REPLACED BY AUDIT AND VERBOSE MODULE NAMES-->",
                 sb.toString());
@@ -226,6 +262,12 @@ private static void reconfig() throws IOException {
         properties.put("dump_roll_num", dumpRollNum);
         properties.put("dump_log_delete_age", dumpDeleteAge);
 
+        properties.put("big_query_log_dir", bigQueryLogDir);
+        properties.put("big_query_file_pattern", bigQueryLogRollPattern);
+        properties.put("big_query_roll_maxsize", bigQueryLogRollMaxSize);
+        properties.put("big_query_roll_num", bigQueryLogRollNum);
+        properties.put("big_query_log_delete_age", bigQueryLogDeleteAge);
+
         strSub = new StrSubstitutor(new Interpolator(properties));
         newXmlConfTemplate = strSub.replace(newXmlConfTemplate);
 
@@ -269,6 +311,7 @@ public static synchronized void initLogging() throws IOException {
         verboseModules = Config.sys_log_verbose_modules;
         auditModules = Config.audit_log_modules;
         dumpModules = Config.dump_log_modules;
+        bigQueryModules = Config.big_query_log_modules;
         reconfig();
     }
 
diff --git a/fe/fe-core/src/main/java/com/starrocks/plugin/AuditEvent.java b/fe/fe-core/src/main/java/com/starrocks/plugin/AuditEvent.java
index 29516f66b6c35..4858c3b8a81f5 100644
--- a/fe/fe-core/src/main/java/com/starrocks/plugin/AuditEvent.java
+++ b/fe/fe-core/src/main/java/com/starrocks/plugin/AuditEvent.java
@@ -102,6 +102,12 @@ public enum EventType {
     public double planMemCosts = 0.0;
     @AuditField(value = "PendingTimeMs")
     public long pendingTimeMs = 0;
+    @AuditField(value = "BigQueryLogCPUSecondThreshold")
+    public long bigQueryLogCPUSecondThreshold = -1;
+    @AuditField(value = "BigQueryLogScanBytesThreshold")
+    public long bigQueryLogScanBytesThreshold = -1;
+    @AuditField(value = "BigQueryLogScanRowsThreshold")
+    public long bigQueryLogScanRowsThreshold = -1;
 
     public static class AuditEventBuilder {
 
@@ -242,6 +248,21 @@ public AuditEventBuilder setPendingTimeMs(long pendingTimeMs) {
             return this;
         }
 
+        public AuditEventBuilder setBigQueryLogCPUSecondThreshold(long bigQueryLogCPUSecondThreshold) {
+            auditEvent.bigQueryLogCPUSecondThreshold = bigQueryLogCPUSecondThreshold;
+            return this;
+        }
+
+        public AuditEventBuilder setBigQueryLogScanBytesThreshold(long bigQueryLogScanBytesThreshold) {
+            auditEvent.bigQueryLogScanBytesThreshold = bigQueryLogScanBytesThreshold;
+            return this;
+        }
+
+        public AuditEventBuilder setBigQueryLogScanRowsThreshold(long bigQueryLogScanRowsThreshold) {
+            auditEvent.bigQueryLogScanRowsThreshold = bigQueryLogScanRowsThreshold;
+            return this;
+        }
+
         public AuditEvent build() {
             return this.auditEvent;
         }
diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/AuditLogBuilder.java b/fe/fe-core/src/main/java/com/starrocks/qe/AuditLogBuilder.java
index e668d499d5468..c1975ef9a74d2 100644
--- a/fe/fe-core/src/main/java/com/starrocks/qe/AuditLogBuilder.java
+++ b/fe/fe-core/src/main/java/com/starrocks/qe/AuditLogBuilder.java
@@ -77,6 +77,14 @@ public void exec(AuditEvent event) {
                     continue;
                 }
 
+                // fields related to big queries are not written into audit log by default,
+                // they will be written into big query log.
+                if (af.value().equals("BigQueryLogCPUSecondThreshold") ||
+                        af.value().equals("BigQueryLogScanBytesThreshold") ||
+                        af.value().equals("BigQueryLogScanRowsThreshold")) {
+                    continue;
+                }
+
                 if (af.value().equals("Time")) {
                     queryTime = (long) f.get(event);
                 }
@@ -89,8 +97,30 @@ public void exec(AuditEvent event) {
             if (queryTime > Config.qe_slow_log_ms) {
                 AuditLog.getSlowAudit().log(auditLog);
             }
+
+            if (isBigQuery(event)) {
+                sb.append("|bigQueryLogCPUSecondThreshold=").append(event.bigQueryLogCPUSecondThreshold);
+                sb.append("|bigQueryLogScanBytesThreshold=").append(event.bigQueryLogScanBytesThreshold);
+                sb.append("|bigQueryLogScanRowsThreshold=").append(event.bigQueryLogScanRowsThreshold);
+                String bigQueryLog = sb.toString();
+                AuditLog.getBigQueryAudit().log(bigQueryLog);
+            }
         } catch (Exception e) {
             LOG.debug("failed to process audit event", e);
         }
     }
+
+    private boolean isBigQuery(AuditEvent event) {
+        if (event.bigQueryLogCPUSecondThreshold >= 0 &&
+                event.cpuCostNs > event.bigQueryLogCPUSecondThreshold * 1000000000L) {
+            return true;
+        }
+        if (event.bigQueryLogScanBytesThreshold >= 0 && event.scanBytes > event.bigQueryLogScanBytesThreshold) {
+            return true;
+        }
+        if (event.bigQueryLogScanRowsThreshold >= 0 && event.scanRows > event.bigQueryLogScanRowsThreshold) {
+            return true;
+        }
+        return false;
+    }
 }
diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/ConnectProcessor.java b/fe/fe-core/src/main/java/com/starrocks/qe/ConnectProcessor.java
index 66754f786d1d1..6f20009e57453 100644
--- a/fe/fe-core/src/main/java/com/starrocks/qe/ConnectProcessor.java
+++ b/fe/fe-core/src/main/java/com/starrocks/qe/ConnectProcessor.java
@@ -171,6 +171,14 @@ public void auditAfterExec(String origStmt, StatementBase parsedStmt, PQueryStat
                 }
             }
             ctx.getAuditEventBuilder().setIsQuery(true);
+            if (ctx.getSessionVariable().isEnableBigQueryLog()) {
+                ctx.getAuditEventBuilder().setBigQueryLogCPUSecondThreshold(
+                        ctx.getSessionVariable().getBigQueryLogCPUSecondThreshold());
+                ctx.getAuditEventBuilder().setBigQueryLogScanBytesThreshold(
+                        ctx.getSessionVariable().getBigQueryLogScanBytesThreshold());
+                ctx.getAuditEventBuilder().setBigQueryLogScanRowsThreshold(
+                        ctx.getSessionVariable().getBigQueryLogScanRowsThreshold());
+            }
         } else {
             ctx.getAuditEventBuilder().setIsQuery(false);
         }
diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/SessionVariable.java b/fe/fe-core/src/main/java/com/starrocks/qe/SessionVariable.java
index 54ffa930079cf..be92304cd02b9 100644
--- a/fe/fe-core/src/main/java/com/starrocks/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/com/starrocks/qe/SessionVariable.java
@@ -305,6 +305,11 @@ public class SessionVariable implements Serializable, Writable, Cloneable {
     public static final String ENABLE_MATERIALIZED_VIEW_UNION_REWRITE = "enable_materialized_view_union_rewrite";
     public static final String ENABLE_RULE_BASED_MATERIALIZED_VIEW_REWRITE = "enable_rule_based_materialized_view_rewrite";
 
+    public static final String ENABLE_BIG_QUERY_LOG = "enable_big_query_log";
+    public static final String BIG_QUERY_LOG_CPU_SECOND_THRESHOLD = "big_query_log_cpu_second_threshold";
+    public static final String BIG_QUERY_LOG_SCAN_BYTES_THRESHOLD = "big_query_log_scan_bytes_threshold";
+    public static final String BIG_QUERY_LOG_SCAN_ROWS_THRESHOLD = "big_query_log_scan_rows_threshold";
+
     public static final List<String> DEPRECATED_VARIABLES = ImmutableList.<String>builder()
             .add(CODEGEN_LEVEL)
             .add(ENABLE_SPILLING)
@@ -753,6 +758,26 @@ public boolean getUseScanBlockCache() {
     @VarAttr(name = ENABLE_RULE_BASED_MATERIALIZED_VIEW_REWRITE)
     private boolean enableRuleBasedMaterializedViewRewrite = true;
 
+    // if enable_big_query_log = true and cpu/io cost of a query exceeds the related threshold,
+    // the information will be written to the big query log
+    @VarAttr(name = ENABLE_BIG_QUERY_LOG)
+    private boolean enableBigQueryLog = true;
+    // the value is set for testing,
+    // if a query needs to perform 10s for computing tasks at full load on three 16-core machines,
+    // we treat it as a big query, so set this value to 480(10 * 16 * 3).
+    // Users need to set up according to their own scenario.
+    @VarAttr(name = BIG_QUERY_LOG_CPU_SECOND_THRESHOLD)
+    private long bigQueryLogCPUSecondThreshold = 480;
+    // the value is set for testing, if a query needs to scan more than 10GB of data, we treat it as a big query.
+    // Users need to set up according to their own scenario.
+    @VarAttr(name = BIG_QUERY_LOG_SCAN_BYTES_THRESHOLD)
+    private long bigQueryLogScanBytesThreshold = 1024L * 1024 * 1024 * 10;
+    // the value is set for testing, if a query need to scan more than 1 billion rows of data,
+    // we treat it as a big query.
+    // Users need to set up according to their own scenario.
+    @VarAttr(name = BIG_QUERY_LOG_SCAN_ROWS_THRESHOLD)
+    private long bigQueryLogScanRowsThreshold = 1000000000L;
+
     public boolean getEnablePopulateBlockCache() {
         return enablePopulateBlockCache;
     }
@@ -1421,6 +1446,38 @@ public void setEnableRuleBasedMaterializedViewRewrite(boolean enableRuleBasedMat
         this.enableRuleBasedMaterializedViewRewrite = enableRuleBasedMaterializedViewRewrite;
     }
 
+    public boolean isEnableBigQueryLog() {
+        return enableBigQueryLog;
+    }
+
+    public void setEnableBigQueryLog(boolean enableBigQueryLog) {
+        this.enableBigQueryLog = enableBigQueryLog;
+    }
+
+    public long getBigQueryLogCPUSecondThreshold() {
+        return this.bigQueryLogCPUSecondThreshold;
+    }
+
+    public void setBigQueryLogCpuSecondThreshold(long bigQueryLogCPUSecondThreshold) {
+        this.bigQueryLogCPUSecondThreshold = bigQueryLogCPUSecondThreshold;
+    }
+
+    public long getBigQueryLogScanBytesThreshold() {
+        return bigQueryLogScanBytesThreshold;
+    }
+
+    public void setBigQueryLogScanBytesThreshold(long bigQueryLogScanBytesThreshold) {
+        this.bigQueryLogScanBytesThreshold = bigQueryLogScanBytesThreshold;
+    }
+
+    public long getBigQueryLogScanRowsThreshold() {
+        return bigQueryLogScanRowsThreshold;
+    }
+
+    public void setBigQueryLogScanRowsThreshold(long bigQueryLogScanRowsThreshold) {
+        this.bigQueryLogScanRowsThreshold = bigQueryLogScanRowsThreshold;
+    }
+
     // Serialize to thrift object
     // used for rest api
     public TQueryOptions toThrift() {