This is an automated email from the ASF dual-hosted git repository.
deardeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new ae61e64702e [refactor](fe) Extract forward handling and bound the
group commit BE cache (#65607)
ae61e64702e is described below
commit ae61e64702e051306418cd86a0375b291f2f5308
Author: deardeng <[email protected]>
AuthorDate: Wed Jul 15 16:19:51 2026 +0800
[refactor](fe) Extract forward handling and bound the group commit BE cache
(#65607)
Problem Summary:
Pure refactor and cleanup in the FE forward / group commit path. No SQL,
protocol
or scheduling semantics change.
1. **`FrontendServiceImpl.forward()` is hard to follow and leaks on the
error path.**
The method inlined requester validation, four shortcut responses,
context and
processor construction, proxy-query registration and execution in one
body. It is
now split into intent-revealing helpers (validate / shortcut / context /
processor
/ execute).
`ConnectContext.remove()` and the proxy-query clear callback used to run
only on
the success path — if `proxyExecute()` threw, the thread-local context
and the
`proxyQueryIdToConnCtx` entry were both left behind. They now run in a
`finally`
block.
The two group commit shortcuts answer without writing a journal, so they
set
`maxJournalId` to `0` explicitly instead of relying on the default value
of that
required field. This does not change the wire format — Thrift always
writes
required primitives — it just makes the intent readable.
2. **`MasterOpExecutor` waited on a journal id that is never produced.**
`getGroupCommitLoadBeId()` and `updateLoadData()` called
`waitOnReplaying()`, but
the master answers both without writing a journal, so the result carries
no journal
id and the call degenerates into `waitOn(0)`. It was a no-op that logged
one INFO
line per request on a hot path. Removed.
3. **`GroupCommitManager` leaked one entry per group-commit table,
forever.**
`tableToBeMap` (table -> BE id) and `tableToPressureMap` (table ->
pressure counter)
were plain maps with no removal, expiry or drop-table cleanup path, so a
workload
that group-commit-loads many short-lived tables grew both maps without
bound. Both
are now bounded caches with the same limits. Keys and BE selection logic
are
unchanged.
Reading the pressure counter no longer does `containsKey()` followed by
`get()`,
which could NPE if the entry was evicted between the two calls.
4. **`FrontendHbResponse.toString()` printed the wrong field.**
It printed `processUUID` under the `festartTime` label and never printed
`feStartTime`.
---
.../org/apache/doris/load/GroupCommitManager.java | 58 ++++----
.../java/org/apache/doris/qe/MasterOpExecutor.java | 4 +-
.../apache/doris/service/FrontendServiceImpl.java | 153 +++++++++++++--------
.../apache/doris/system/FrontendHbResponse.java | 3 +-
4 files changed, 136 insertions(+), 82 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java
b/fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java
index 534d38a09ee..73cb7f2c4f4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java
@@ -40,6 +40,8 @@ import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TStatusCode;
import com.google.common.base.Strings;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.Nullable;
@@ -48,10 +50,9 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
-import java.util.Map;
import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
public class GroupCommitManager {
@@ -60,10 +61,19 @@ public class GroupCommitManager {
private Set<Long> blockedTableIds = new HashSet<>();
- // Encoded <Cluster and Table id> to BE id map. Only for group commit.
- private final Map<String, Long> tableToBeMap = new ConcurrentHashMap<>();
+ // Encoded <Cluster and Table id> to BE id cache. Only for group commit.
+ // Bounded so that dropped tables cannot keep their entries alive forever.
+ private final Cache<String, Long> tableToBeMap = CacheBuilder.newBuilder()
+ .maximumSize(10000)
+ .expireAfterAccess(1, TimeUnit.HOURS)
+ .build();
// Table id to pressure map. Only for group commit.
- private final Map<Long, SlidingWindowCounter> tableToPressureMap = new
ConcurrentHashMap<>();
+ // Bounded like tableToBeMap above: the two hold one entry per
group-commit table and are only
+ // meaningful together, so a dropped table must not keep either of them
alive.
+ private final Cache<Long, SlidingWindowCounter> tableToPressureMap =
CacheBuilder.newBuilder()
+ .maximumSize(10000)
+ .expireAfterAccess(1, TimeUnit.HOURS)
+ .build();
public boolean isBlock(long tableId) {
return blockedTableIds.contains(tableId);
@@ -277,7 +287,7 @@ public class GroupCommitManager {
throws DdlException, LoadException {
if (LOG.isDebugEnabled()) {
LOG.debug("cloud group commit select be info, tableToBeMap {},
tablePressureMap {}",
- tableToBeMap.toString(), tableToPressureMap.toString());
+ tableToBeMap.asMap().toString(),
tableToPressureMap.asMap().toString());
}
if (Strings.isNullOrEmpty(cluster)) {
ErrorReport.reportDdlException(ErrorCode.ERR_NO_CLUSTER_ERROR);
@@ -309,8 +319,8 @@ public class GroupCommitManager {
private long selectBackendForLocalGroupCommitInternal(long tableId) throws
LoadException {
if (LOG.isDebugEnabled()) {
- LOG.debug("group commit select be info, tableToBeMap {},
tablePressureMap {}", tableToBeMap.toString(),
- tableToPressureMap.toString());
+ LOG.debug("group commit select be info, tableToBeMap {},
tablePressureMap {}",
+ tableToBeMap.asMap().toString(),
tableToPressureMap.asMap().toString());
}
Long cachedBackendId = getCachedBackend(null, tableId);
if (cachedBackendId != null) {
@@ -345,26 +355,25 @@ public class GroupCommitManager {
@Nullable
private Long getCachedBackend(String cluster, long tableId) {
OlapTable table = (OlapTable)
Env.getCurrentEnv().getInternalCatalog().getTableByTableId(tableId);
- if (tableToBeMap.containsKey(encode(cluster, tableId))) {
- if (tableToPressureMap.get(tableId) == null) {
+ String cacheKey = encode(cluster, tableId);
+ // There are multiple threads getting cached backends for the same
table.
+ // Maybe one thread removes the tableId from the tableToBeMap.
+ // Another thread gets the same tableId but can not find this tableId.
+ // So another thread needs to get the random backend.
+ Long backendId = tableToBeMap.getIfPresent(cacheKey);
+ if (backendId != null) {
+ SlidingWindowCounter pressure =
tableToPressureMap.getIfPresent(tableId);
+ if (pressure == null) {
return null;
- } else if (tableToPressureMap.get(tableId).get() <
table.getGroupCommitDataBytes()) {
- // There are multiple threads getting cached backends for the
same table.
- // Maybe one thread removes the tableId from the tableToBeMap.
- // Another thread gets the same tableId but can not find this
tableId.
- // So another thread needs to get the random backend.
- Long backendId = tableToBeMap.get(encode(cluster, tableId));
- if (backendId == null) {
- return null;
- }
+ } else if (pressure.get() < table.getGroupCommitDataBytes()) {
Backend backend =
Env.getCurrentSystemInfo().getBackend(backendId);
if (isBackendAvailable(backend, cluster)) {
return backend.getId();
} else {
- tableToBeMap.remove(encode(cluster, tableId));
+ tableToBeMap.invalidate(cacheKey);
}
} else {
- tableToBeMap.remove(encode(cluster, tableId));
+ tableToBeMap.invalidate(cacheKey);
}
}
return null;
@@ -426,11 +435,12 @@ public class GroupCommitManager {
}
private void updateLoadDataInternal(long tableId, long receiveData) {
- if (tableToPressureMap.containsKey(tableId)) {
- tableToPressureMap.get(tableId).add(receiveData);
+ SlidingWindowCounter pressure =
tableToPressureMap.getIfPresent(tableId);
+ if (pressure != null) {
+ pressure.add(receiveData);
if (LOG.isDebugEnabled()) {
LOG.debug("Update load data for table {}, receiveData {},
tablePressureMap {}", tableId, receiveData,
- tableToPressureMap.toString());
+ tableToPressureMap.asMap().toString());
}
} else if (LOG.isDebugEnabled()) {
LOG.debug("can not find table id {}", tableId);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/MasterOpExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/MasterOpExecutor.java
index 067db673049..892dd3cfd73 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/MasterOpExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/MasterOpExecutor.java
@@ -75,15 +75,15 @@ public class MasterOpExecutor extends FEOpExecutor {
waitOnReplaying();
}
+ // The master handles the group commit shortcuts without writing a
journal, so the result carries
+ // no journal id. Waiting on journal 0 is a no-op that only logs one line
per request.
public long getGroupCommitLoadBeId(long tableId, String cluster) throws
Exception {
result = forward(buildGetGroupCommitLoadBeIdParmas(tableId, cluster));
- waitOnReplaying();
return result.groupCommitLoadBeId;
}
public void updateLoadData(long tableId, long receiveData) throws
Exception {
result = forward(buildUpdateLoadDataParams(tableId, receiveData));
- waitOnReplaying();
}
private TMasterOpRequest buildSyncJournalParams() {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index afd0ed46b2a..ef405f907a1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -1152,61 +1152,97 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
@Override
public TMasterOpResult forward(TMasterOpRequest params) throws TException {
+ validateForwardRequester(params);
+ TMasterOpResult shortcut = handleForwardShortcut(params);
+ if (shortcut != null) {
+ return shortcut;
+ }
+ logForwardRequest(params);
+ ConnectContext context = createForwardContext(params);
+ ConnectProcessor processor = createForwardProcessor(context);
+ Runnable clearCallback = registerProxyQuery(params, context);
+ try {
+ return executeForward(params, context, processor);
+ } finally {
+ ConnectContext.remove();
+ clearCallback.run();
+ }
+ }
+
+ private void validateForwardRequester(TMasterOpRequest params) throws
TException {
Frontend fe =
Env.getCurrentEnv().checkFeExist(params.getClientNodeHost(),
params.getClientNodePort());
- if (fe == null) {
- LOG.warn("reject request from invalid host. client: {}",
params.getClientNodeHost());
- throw new TException("request from invalid host was rejected.");
+ if (fe != null) {
+ return;
}
+ LOG.warn("reject request from invalid host. client: {}",
params.getClientNodeHost());
+ throw new TException("request from invalid host was rejected.");
+ }
+
+ private TMasterOpResult handleForwardShortcut(TMasterOpRequest params)
throws TException {
if (params.isSyncJournalOnly()) {
- final TMasterOpResult result = new TMasterOpResult();
- result.setMaxJournalId(Env.getCurrentEnv().getMaxJournalId());
- // just make the protocol happy
- result.setPacket("".getBytes());
- return result;
+ return createForwardResultWithJournalSync();
}
if (params.getGroupCommitInfo() != null &&
params.getGroupCommitInfo().isGetGroupCommitLoadBeId()) {
- final TGroupCommitInfo info = params.getGroupCommitInfo();
- final TMasterOpResult result = new TMasterOpResult();
- try {
-
result.setGroupCommitLoadBeId(Env.getCurrentEnv().getGroupCommitManager()
-
.selectBackendForGroupCommitInternal(info.groupCommitLoadTableId,
info.cluster));
- } catch (LoadException | DdlException e) {
- throw new TException(e.getMessage());
- }
- // just make the protocol happy
- result.setPacket("".getBytes());
- return result;
+ return handleGroupCommitLoadBeId(params.getGroupCommitInfo());
}
if (params.getGroupCommitInfo() != null &&
params.getGroupCommitInfo().isUpdateLoadData()) {
- final TGroupCommitInfo info = params.getGroupCommitInfo();
- final TMasterOpResult result = new TMasterOpResult();
Env.getCurrentEnv().getGroupCommitManager()
- .updateLoadData(info.tableId, info.receiveData);
- // just make the protocol happy
- result.setPacket("".getBytes());
- return result;
+ .updateLoadData(params.getGroupCommitInfo().tableId,
params.getGroupCommitInfo().receiveData);
+ return createForwardResultWithoutJournalSync();
}
- if (params.isSetCancelQeury() && params.isCancelQeury()) {
- if (!params.isSetQueryId()) {
- throw new TException("a query id is needed to cancel a query");
- }
- TUniqueId queryId = params.getQueryId();
- ConnectContext ctx = proxyQueryIdToConnCtx.get(queryId);
- if (ctx != null) {
- ctx.cancelQuery(new Status(TStatusCode.CANCELLED, "cancel
query by forward request."));
- }
- final TMasterOpResult result = new TMasterOpResult();
- result.setStatusCode(0);
- result.setMaxJournalId(Env.getCurrentEnv().getMaxJournalId());
- // just make the protocol happy
- result.setPacket("".getBytes());
- return result;
+ if (!params.isSetCancelQeury() || !params.isCancelQeury()) {
+ return null;
+ }
+ return handleForwardCancel(params);
+ }
+
+ private TMasterOpResult createForwardResultWithJournalSync() {
+ TMasterOpResult result = new TMasterOpResult();
+ result.setMaxJournalId(Env.getCurrentEnv().getMaxJournalId());
+ result.setPacket("".getBytes());
+ return result;
+ }
+
+ private TMasterOpResult createForwardResultWithoutJournalSync() {
+ TMasterOpResult result = new TMasterOpResult();
+ // Group commit shortcuts update master memory without producing a
journal id. Say so explicitly
+ // instead of relying on the default value of the required
maxJournalId field.
+ result.setMaxJournalId(0L);
+ result.setPacket("".getBytes());
+ return result;
+ }
+
+ private TMasterOpResult handleGroupCommitLoadBeId(TGroupCommitInfo info)
throws TException {
+ TMasterOpResult result = createForwardResultWithoutJournalSync();
+ try {
+
result.setGroupCommitLoadBeId(Env.getCurrentEnv().getGroupCommitManager()
+
.selectBackendForGroupCommitInternal(info.groupCommitLoadTableId,
info.cluster));
+ } catch (LoadException | DdlException e) {
+ throw new TException(e.getMessage());
}
+ return result;
+ }
+
+ private TMasterOpResult handleForwardCancel(TMasterOpRequest params)
throws TException {
+ if (!params.isSetQueryId()) {
+ throw new TException("a query id is needed to cancel a query");
+ }
+ ConnectContext context =
proxyQueryIdToConnCtx.get(params.getQueryId());
+ if (context != null) {
+ context.cancelQuery(new Status(TStatusCode.CANCELLED, "cancel
query by forward request."));
+ }
+ TMasterOpResult result = createForwardResultWithJournalSync();
+ result.setStatusCode(0);
+ return result;
+ }
- // add this log so that we can track this stmt
+ private void logForwardRequest(TMasterOpRequest params) {
if (LOG.isDebugEnabled()) {
LOG.debug("receive forwarded stmt {} from FE: {}",
params.getStmtId(), params.getClientNodeHost());
}
+ }
+
+ private ConnectContext createForwardContext(TMasterOpRequest params) {
ConnectContext context = new ConnectContext(null, true,
params.getSessionId());
// Set current connected FE to the client address, so that we can know
where
// this request come from.
@@ -1214,28 +1250,35 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
if (Config.isCloudMode() &&
!Strings.isNullOrEmpty(params.getCloudCluster())) {
context.setCloudCluster(params.getCloudCluster());
}
+ return context;
+ }
- ConnectProcessor processor = null;
+ private ConnectProcessor createForwardProcessor(ConnectContext context)
throws TException {
if (context.getConnectType().equals(ConnectType.MYSQL)) {
- processor = new MysqlConnectProcessor(context);
- } else if
(context.getConnectType().equals(ConnectType.ARROW_FLIGHT_SQL)) {
- processor = new FlightSqlConnectProcessor(context);
- } else {
- throw new TException("unknown ConnectType: " +
context.getConnectType());
+ return new MysqlConnectProcessor(context);
+ }
+ if (context.getConnectType().equals(ConnectType.ARROW_FLIGHT_SQL)) {
+ return new FlightSqlConnectProcessor(context);
}
- Runnable clearCallback = () -> {};
- if (params.isSetQueryId()) {
- proxyQueryIdToConnCtx.put(params.getQueryId(), context);
- clearCallback = () ->
proxyQueryIdToConnCtx.remove(params.getQueryId());
+ throw new TException("unknown ConnectType: " +
context.getConnectType());
+ }
+
+ private Runnable registerProxyQuery(TMasterOpRequest params,
ConnectContext context) {
+ if (!params.isSetQueryId()) {
+ return () -> {};
}
+ proxyQueryIdToConnCtx.put(params.getQueryId(), context);
+ return () -> proxyQueryIdToConnCtx.remove(params.getQueryId());
+ }
+
+ private TMasterOpResult executeForward(TMasterOpRequest params,
ConnectContext context,
+ ConnectProcessor processor) throws TException {
TMasterOpResult result = processor.proxyExecute(params);
if
(QueryState.MysqlStateType.ERR.name().equalsIgnoreCase(result.getStatus())) {
context.getState().setError(result.getStatus());
- } else {
- context.getState().setOk();
+ return result;
}
- ConnectContext.remove();
- clearCallback.run();
+ context.getState().setOk();
return result;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/system/FrontendHbResponse.java
b/fe/fe-core/src/main/java/org/apache/doris/system/FrontendHbResponse.java
index 0bbc72008dd..be8eecfc193 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/system/FrontendHbResponse.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/system/FrontendHbResponse.java
@@ -121,7 +121,8 @@ public class FrontendHbResponse extends HeartbeatResponse
implements Writable {
sb.append(", rpcPort: ").append(rpcPort);
sb.append(", arrowFlightSqlPort: ").append(arrowFlightSqlPort);
sb.append(", replayedJournalId: ").append(replayedJournalId);
- sb.append(", festartTime: ").append(processUUID);
+ sb.append(", feStartTime: ").append(feStartTime);
+ sb.append(", processUUID: ").append(processUUID);
return sb.toString();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
