[improve](group commit) set internal group commit timeout (#41404)

The internal group commit timeout now use the default value of load,
which is 14400 second, 4 hour.
It should not use it, because the group commit interval is generally a
small value.

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -577,6 +577,11 @@ public class Config extends ConfigBase {
             "Default commit data bytes for group commit"})
     public static int group_commit_data_bytes_default_value = 134217728;
 
+    @ConfField(mutable = true, masterOnly = true, description = {
+            "内部攒批的超时时间为table的group_commit_interval_ms的倍数",
+            "The internal group commit timeout is the multiple of table's group_commit_interval_ms"})
+    public static int group_commit_timeout_multipler = 10;
+
     @ConfField(mutable = true, masterOnly = true, description = {"Stream load 的默认超时时间，单位是秒。",
             "Default timeout for stream load job, in seconds."})
     public static int stream_load_default_timeout_second = 86400 * 3; // 3days

fe/fe-core/src/main/java/org/apache/doris/tablefunction/GroupCommitTableValuedFunction.java

@@ -25,9 +25,11 @@
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.Table;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
 import org.apache.doris.planner.GroupCommitScanNode;
 import org.apache.doris.planner.PlanNodeId;
 import org.apache.doris.planner.ScanNode;
+import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.thrift.TFileType;
 
 import java.util.ArrayList;
@@ -66,6 +68,13 @@ public List<Column> getTableColumns() throws AnalysisException {
             throw new AnalysisException("Only support OLAP table, but table type of table_id "
                     + tableId + " is " + table.getType());
         }
+        if (Config.group_commit_timeout_multipler > 0) {
+            int timeoutS = Math.max((int) (((OlapTable) table).getGroupCommitIntervalMs() / 1000.0
+                    * Config.group_commit_timeout_multipler), 600);
+            ConnectContext.get().getSessionVariable().setInsertTimeoutS(timeoutS);
+            ConnectContext.get().getSessionVariable().setQueryTimeoutS(timeoutS);
+        }
+
         List<Column> tableColumns = table.getBaseSchema(true);
         for (int i = 1; i <= tableColumns.size(); i++) {
             fileColumns.add(new Column("c" + i, tableColumns.get(i - 1).getType(), true));