diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java index 9811eef28b6337..1b0b3f77b2b3da 100644 --- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java +++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java @@ -3137,4 +3137,9 @@ public static int metaServiceRpcRetryTimes() { @ConfField(mutable = true, description = {"元数据同步是否开启安全模式", "Is metadata synchronization enabled in safe mode"}) public static boolean meta_helper_security_mode = false; + + @ConfField(description = {"用于测试,强制将所有的查询forward到master以验证forward query的行为", + "For testing purposes, all queries are forcibly forwarded to the master to verify" + + "the behavior of forwarding queries."}) + public static boolean force_forward_all_queries = false; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java index 02aef153e37f93..2340aa37aebb55 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java @@ -560,11 +560,8 @@ public void finalizeCommand() throws IOException { && ctx.getState().getStateType() != QueryState.MysqlStateType.ERR) { ShowResultSet resultSet = executor.getShowResultSet(); if (resultSet == null) { - if (executor.sendProxyQueryResult()) { - packet = getResultPacket(); - } else { - packet = executor.getOutputPacket(); - } + executor.sendProxyQueryResult(); + packet = executor.getOutputPacket(); } else { executor.sendResultSet(resultSet); packet = getResultPacket(); @@ -724,7 +721,12 @@ public TMasterOpResult proxyExecute(TMasterOpRequest request) throws TException if (ctx.getState().getStateType() == MysqlStateType.OK) { result.setStatusCode(0); } else { - result.setStatusCode(ctx.getState().getErrorCode().getCode()); + ErrorCode errorCode = ctx.getState().getErrorCode(); + if (errorCode != null) { + result.setStatusCode(errorCode.getCode()); + } else { + result.setStatusCode(ErrorCode.ERR_UNKNOWN_ERROR.getCode()); + } result.setErrMessage(ctx.getState().getErrorMessage()); } if (request.isSetTxnLoadInfo()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java index 16600536aac1b9..234d5f0e610780 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java @@ -283,6 +283,7 @@ public class StmtExecutor { private boolean isHandleQueryInFe = false; // The profile of this execution private final Profile profile; + private Boolean isForwardedToMaster = null; // The result schema if "dry_run_query" is true. // Only one column to indicate the real return row numbers. @@ -423,6 +424,13 @@ public void setPlanner(Planner planner) { } public boolean isForwardToMaster() { + if (isForwardedToMaster == null) { + isForwardedToMaster = shouldForwardToMaster(); + } + return isForwardedToMaster; + } + + private boolean shouldForwardToMaster() { if (Env.getCurrentEnv().isMaster()) { return false; } @@ -433,7 +441,7 @@ public boolean isForwardToMaster() { // this is a query stmt, but this non-master FE can not read, forward it to master if (isQuery() && !Env.getCurrentEnv().isMaster() - && (!Env.getCurrentEnv().canRead() || debugForwardAllQueries())) { + && (!Env.getCurrentEnv().canRead() || debugForwardAllQueries() || Config.force_forward_all_queries)) { return true; } @@ -446,7 +454,7 @@ public boolean isForwardToMaster() { private boolean debugForwardAllQueries() { DebugPoint debugPoint = DebugPointUtil.getDebugPoint("StmtExecutor.forward_all_queries"); - return debugPoint != null && debugPoint.param("forwardAllQueries", true); + return debugPoint != null && debugPoint.param("forwardAllQueries", false); } public ByteBuffer getOutputPacket() { @@ -3618,17 +3626,13 @@ public List getProxyQueryResultBufList() { return ((ProxyMysqlChannel) context.getMysqlChannel()).getProxyResultBufferList(); } - public boolean sendProxyQueryResult() throws IOException { + public void sendProxyQueryResult() throws IOException { if (masterOpExecutor == null) { - return false; + return; } List queryResultBufList = masterOpExecutor.getQueryResultBufList(); - if (queryResultBufList.isEmpty()) { - return false; - } for (ByteBuffer byteBuffer : queryResultBufList) { context.getMysqlChannel().sendOnePacket(byteBuffer); } - return true; } } diff --git a/regression-test/suites/query_p0/test_forward_qeury.groovy b/regression-test/suites/query_p0/test_forward_qeury.groovy index 798e8865ca5074..d4761c835a26e0 100644 --- a/regression-test/suites/query_p0/test_forward_qeury.groovy +++ b/regression-test/suites/query_p0/test_forward_qeury.groovy @@ -41,8 +41,9 @@ suite("test_forward_query", 'docker') { sql """ INSERT INTO ${tbl} VALUES(1);""" - cluster.injectDebugPoints(NodeType.FE, ['StmtExecutor.forward_all_queries' : [forwardAllQueries:true]]) + cluster.injectDebugPoints(NodeType.FE, ['StmtExecutor.forward_all_queries' : [forwardAllQueries:true, execute:1]]) - sql """ SELECT * FROM ${tbl} """ + def ret = sql """ SELECT * FROM ${tbl} """ + assertEquals(ret[0][0], 1) } }