From 8d61a2cfdd4224ccda108bd85a1166cf88f687f0 Mon Sep 17 00:00:00 2001 From: wyb Date: Fri, 20 Sep 2024 13:30:27 +0800 Subject: [PATCH] Add ut Signed-off-by: wyb --- .../java/com/starrocks/qe/ConnectContext.java | 2 +- .../java/com/starrocks/qe/ResultReceiver.java | 8 ++-- .../java/com/starrocks/qe/StmtExecutor.java | 2 +- .../com/starrocks/qe/StmtExecutorTest.java | 38 +++++++++++++++++++ 4 files changed, 44 insertions(+), 6 deletions(-) diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/ConnectContext.java b/fe/fe-core/src/main/java/com/starrocks/qe/ConnectContext.java index 12fc20bf3372ba..fa4047423effae 100644 --- a/fe/fe-core/src/main/java/com/starrocks/qe/ConnectContext.java +++ b/fe/fe-core/src/main/java/com/starrocks/qe/ConnectContext.java @@ -956,7 +956,7 @@ public void checkTimeout(long now) { killFlag = true; String suggestedMsg = String.format("please increase the '%s' session variable", - isExecLoadType() ? "insert_timeout" : "query_timeout"); + isExecLoadType() ? SessionVariable.INSERT_TIMEOUT : SessionVariable.QUERY_TIMEOUT); errMsg = ErrorCode.ERR_TIMEOUT.formatErrorMsg(getExecType(), timeoutSecond, suggestedMsg); } } diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/ResultReceiver.java b/fe/fe-core/src/main/java/com/starrocks/qe/ResultReceiver.java index fc71057298ac01..95f5e1e7d105d0 100644 --- a/fe/fe-core/src/main/java/com/starrocks/qe/ResultReceiver.java +++ b/fe/fe-core/src/main/java/com/starrocks/qe/ResultReceiver.java @@ -141,16 +141,16 @@ public RowBatch getNext(Status status) throws TException { LOG.warn("fetch result execution exception, finstId={}", DebugUtil.printId(finstId), e); if (e.getMessage().contains("time out")) { // if timeout, we set error code to TIMEOUT, and it will not retry querying. - status.setStatus(new Status(TStatusCode.TIMEOUT, ErrorCode.ERR_TIMEOUT.formatErrorMsg( - "Query", timeoutMs / 1000, "please increase the 'query_timeout' session variable"))); + status.setStatus(new Status(TStatusCode.TIMEOUT, ErrorCode.ERR_TIMEOUT.formatErrorMsg("Query", timeoutMs / 1000, + String.format("please increase the '%s' session variable", SessionVariable.QUERY_TIMEOUT)))); } else { status.setRpcStatus(e.getMessage()); SimpleScheduler.addToBlocklist(backendId); } } catch (TimeoutException e) { LOG.warn("fetch result timeout, finstId={}", DebugUtil.printId(finstId), e); - status.setInternalErrorStatus(ErrorCode.ERR_TIMEOUT.formatErrorMsg( - "Query", timeoutMs / 1000, "please increase the 'query_timeout' session variable")); + status.setInternalErrorStatus(ErrorCode.ERR_TIMEOUT.formatErrorMsg("Query", timeoutMs / 1000, + String.format("please increase the '%s' session variable", SessionVariable.QUERY_TIMEOUT))); if (MetricRepo.hasInit) { MetricRepo.COUNTER_QUERY_TIMEOUT.increase(1L); } diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/StmtExecutor.java b/fe/fe-core/src/main/java/com/starrocks/qe/StmtExecutor.java index a24a2c3175a8b9..bc536b0896d3b9 100644 --- a/fe/fe-core/src/main/java/com/starrocks/qe/StmtExecutor.java +++ b/fe/fe-core/src/main/java/com/starrocks/qe/StmtExecutor.java @@ -2247,7 +2247,7 @@ public void handleDMLStmt(ExecPlan execPlan, DmlStmt stmt) throws Exception { "or set parallel_fragment_exec_instance_num to a lower value in session variable"); } else { ErrorReport.reportDdlException(ErrorCode.ERR_TIMEOUT, getExecType(), timeout, - "please increase the 'insert_timeout' session variable"); + String.format("please increase the '%s' session variable"), SessionVariable.INSERT_TIMEOUT); } } } diff --git a/fe/fe-core/src/test/java/com/starrocks/qe/StmtExecutorTest.java b/fe/fe-core/src/test/java/com/starrocks/qe/StmtExecutorTest.java index 762a9c9273777e..0556a22a1fe1a4 100644 --- a/fe/fe-core/src/test/java/com/starrocks/qe/StmtExecutorTest.java +++ b/fe/fe-core/src/test/java/com/starrocks/qe/StmtExecutorTest.java @@ -15,8 +15,10 @@ package com.starrocks.qe; import com.starrocks.server.GlobalStateMgr; +import com.starrocks.sql.ast.StatementBase; import com.starrocks.sql.parser.AstBuilder; import com.starrocks.sql.parser.SqlParser; +import com.starrocks.utframe.UtFrameUtils; import mockit.Expectations; import mockit.Mocked; import org.junit.Assert; @@ -49,4 +51,40 @@ public void testIsForwardToLeader(@Mocked GlobalStateMgr state) { Assert.assertFalse(new StmtExecutor(new ConnectContext(), SqlParser.parseSingleStatement("show frontends", SqlModeHelper.MODE_DEFAULT)).isForwardToLeader()); } + + @Test + public void testExecType() { + ConnectContext ctx = UtFrameUtils.createDefaultCtx(); + ConnectContext.threadLocalInfo.set(ctx); + + StatementBase stmt = SqlParser.parseSingleStatement("select * from t1", SqlModeHelper.MODE_DEFAULT); + StmtExecutor executor = new StmtExecutor(new ConnectContext(), stmt); + Assert.assertEquals("Query", executor.getExecType()); + Assert.assertFalse(executor.isExecLoadType()); + Assert.assertEquals(ConnectContext.get().getSessionVariable().getQueryTimeoutS(), executor.getExecTimeout()); + + stmt = SqlParser.parseSingleStatement("insert into t1 select * from t2", SqlModeHelper.MODE_DEFAULT); + executor = new StmtExecutor(new ConnectContext(), stmt); + Assert.assertEquals("Insert", executor.getExecType()); + Assert.assertTrue(executor.isExecLoadType()); + Assert.assertEquals(ConnectContext.get().getSessionVariable().getInsertTimeoutS(), executor.getExecTimeout()); + + stmt = SqlParser.parseSingleStatement("create table t1 as select * from t2", SqlModeHelper.MODE_DEFAULT); + executor = new StmtExecutor(new ConnectContext(), stmt); + Assert.assertEquals("Insert", executor.getExecType()); + Assert.assertTrue(executor.isExecLoadType()); + Assert.assertEquals(ConnectContext.get().getSessionVariable().getInsertTimeoutS(), executor.getExecTimeout()); + + stmt = SqlParser.parseSingleStatement("update t1 set k1 = 1 where k2 = 1", SqlModeHelper.MODE_DEFAULT); + executor = new StmtExecutor(new ConnectContext(), stmt); + Assert.assertEquals("Update", executor.getExecType()); + Assert.assertTrue(executor.isExecLoadType()); + Assert.assertEquals(ConnectContext.get().getSessionVariable().getInsertTimeoutS(), executor.getExecTimeout()); + + stmt = SqlParser.parseSingleStatement("delete from t1 where k2 = 1", SqlModeHelper.MODE_DEFAULT); + executor = new StmtExecutor(new ConnectContext(), stmt); + Assert.assertEquals("Delete", executor.getExecType()); + Assert.assertTrue(executor.isExecLoadType()); + Assert.assertEquals(ConnectContext.get().getSessionVariable().getInsertTimeoutS(), executor.getExecTimeout()); + } }