For #5169 Add publish timout param when exec insert

docs/en/administrator-guide/variables.md

@@ -74,6 +74,7 @@ Variables that support both session-level and global-level setting include:
 * `parallel_fragment_exec_instance_num`
 * `parallel_exchange_instance_num`
 * `allow_partition_column_nullable`
+* `insert_visible_timeout_ms`
 
 Variables that support only global-level setting include:
 
@@ -364,4 +365,9 @@ Note that the comment must start with /*+ and can only follow the SELECT.
 
 * `allow_partition_column_nullable`
 
-    Whether to allow the partition column to be NULL when creating the table. The default is true, which means NULL is allowed. false means the partition column must be defined as NOT NULL.
+    Whether to allow the partition column to be NULL when creating the table. The default is true, which means NULL is allowed. false means the partition column must be defined as NOT NULL.
+
+* `insert_visible_timeout_ms`
+
+    When execute insert statement, doris will wait for the transaction to commit and visible after the import is completed.
+    This parameter controls the timeout of waiting for transaction to be visible. The default value is 10000, and the minimum value is 1000.

docs/zh-CN/administrator-guide/variables.md

@@ -74,6 +74,7 @@ SET GLOBAL exec_mem_limit = 137438953472
 * `parallel_fragment_exec_instance_num`
 * `parallel_exchange_instance_num`
 * `allow_partition_column_nullable`
+* `insert_visible_timeout_ms`
 
 只支持全局生效的变量包括：
 
@@ -363,4 +364,8 @@ SELECT /*+ SET_VAR(query_timeout = 1) */ sleep(3);
 
 * `allow_partition_column_nullable`
 
-    建表时是否允许分区列为NULL。默认为true，表示允许为NULL。false 表示分区列必须被定义为NOT NULL
+    建表时是否允许分区列为NULL。默认为true，表示允许为NULL。false 表示分区列必须被定义为NOT NULL
+
+* `insert_visible_timeout_ms`
+
+    在执行insert语句时，导入动作(查询和插入)完成后，还需要等待事务提交，使数据可见。此参数控制等待数据可见的超时时间，默认为10000，最小为1000。

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

@@ -415,6 +415,10 @@ public TMasterOpResult proxyExecute(TMasterOpRequest request) {
             ctx.setCurrentUserIdentity(currentUserIdentity);
         }
 
+        if (request.isSetInsertVisibleTimeoutMs()) {
+            ctx.getSessionVariable().setInsertVisibleTimeoutMs(request.getInsertVisibleTimeoutMs());
+        }
+
         if (request.isSetQueryOptions()) {
             TQueryOptions queryOptions = request.getQueryOptions();
             if (queryOptions.isSetMemLimit()) {

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

@@ -93,6 +93,7 @@ private void forward() throws Exception {
         params.setStmtId(ctx.getStmtId());
         params.setEnableStrictMode(ctx.getSessionVariable().getEnableInsertStrict());
         params.setCurrentUserIdent(ctx.getCurrentUserIdentity().toThrift());
+        params.setInsertVisibleTimeoutMs(ctx.getSessionVariable().getInsertVisibleTimeoutMs());
 
         TQueryOptions queryOptions = new TQueryOptions();
         queryOptions.setMemLimit(ctx.getSessionVariable().getMaxExecMemByte());

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

@@ -112,6 +112,14 @@ public class SessionVariable implements Serializable, Writable {
     // when true, the partition column must be set to NOT NULL.
     public static final String ALLOW_PARTITION_COLUMN_NULLABLE = "allow_partition_column_nullable";
 
+    // max ms to wait transaction publish finish when exec insert stmt.
+    public static final String INSERT_VISIBLE_TIMEOUT_MS = "insert_visible_timeout_ms";
+    public static final long DEFAULT_INSERT_VISIBLE_TIMEOUT_MS = 10_000;
+    public static final long MIN_INSERT_VISIBLE_TIMEOUT_MS = 1000; // If user set a very small value, use this value instead.
+
+    @VariableMgr.VarAttr(name = INSERT_VISIBLE_TIMEOUT_MS)
+    private long insertVisibleTimeoutMs = DEFAULT_INSERT_VISIBLE_TIMEOUT_MS;
+
     // max memory used on every backend.
     @VariableMgr.VarAttr(name = EXEC_MEM_LIMIT)
     public long maxExecMemByte = 2147483648L;
@@ -544,6 +552,21 @@ public void setShowHiddenColumns(boolean showHiddenColumns) {
 
     public boolean isAllowPartitionColumnNullable() { return allowPartitionColumnNullable; }
 
+    public long getInsertVisibleTimeoutMs() {
+        if (insertVisibleTimeoutMs < MIN_INSERT_VISIBLE_TIMEOUT_MS) {
+            return MIN_INSERT_VISIBLE_TIMEOUT_MS;
+        } else {
+            return insertVisibleTimeoutMs;
+        }
+    }
+
+    public void setInsertVisibleTimeoutMs(long insertVisibleTimeoutMs) {
+        if (insertVisibleTimeoutMs < MIN_INSERT_VISIBLE_TIMEOUT_MS) {
+            this.insertVisibleTimeoutMs = MIN_INSERT_VISIBLE_TIMEOUT_MS;
+        } else {
+            this.insertVisibleTimeoutMs = insertVisibleTimeoutMs;
+        }
+    }
 
     // Serialize to thrift object
     // used for rest api

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

@@ -897,11 +897,10 @@ private void handleInsertStmt() throws Exception {
                 context.getState().setOk();
                 return;
             }
-
             if (Catalog.getCurrentGlobalTransactionMgr().commitAndPublishTransaction(
                     insertStmt.getDbObj(), insertStmt.getTransactionId(),
                     TabletCommitInfo.fromThrift(coord.getCommitInfos()),
-                    10000)) {
+                    context.getSessionVariable().getInsertVisibleTimeoutMs())) {
                 txnStatus = TransactionStatus.VISIBLE;
                 MetricRepo.COUNTER_LOAD_FINISHED.increase(1L);
             } else {

gensrc/thrift/FrontendService.thrift

@@ -443,6 +443,7 @@ struct TMasterOpRequest {
     15: optional i32 stmtIdx  // the idx of the sql in multi statements
     16: optional PaloInternalService.TQueryOptions query_options
     17: optional Types.TUniqueId query_id // when this is a query, we translate this query id to master
+    18: optional i64 insert_visible_timeout_ms
 }
 
 struct TColumnDefinition {