dataroaring commented on code in PR #35558:
URL: https://github.com/apache/doris/pull/35558#discussion_r1621787075
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -2618,6 +2618,13 @@ public class Config extends ConfigBase {
})
public static boolean enable_advance_next_id = false;
+ // Advance the next id before transferring to the master.
Review Comment:
The comment is wrong.
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -2618,6 +2618,13 @@ public class Config extends ConfigBase {
})
public static boolean enable_advance_next_id = false;
+ // Advance the next id before transferring to the master.
+ @ConfField(description = {
+ "是否采用新的group commit BE选择算法",
+ "Whether to use new group commit BE select strategy."
+ })
+ public static boolean enable_new_group_commit_be_select_strategy = false;
Review Comment:
new is not a good word here.
##########
fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java:
##########
@@ -163,4 +177,76 @@ public long getWalQueueSize(Backend backend,
PGetWalQueueSizeRequest request) {
return size;
}
+ public Backend selectBackendForGroupCommit(long tableId, ConnectContext
context) throws LoadException {
+ if (!Env.getCurrentEnv().isMaster()) {
+ try {
+ long backendId = new
MasterOpExecutor(context).getGroupCommitLoadBeId(tableId);
+ return Env.getCurrentSystemInfo().getBackend(backendId);
+ } catch (Exception e) {
+ throw new LoadException(e.getMessage());
+ }
+ } else {
+ return
Env.getCurrentSystemInfo().getBackend(selectBackendForGroupCommitInternal(tableId));
+ }
+ }
+
+ public long selectBackendForGroupCommitInternal(long tableId) throws
LoadException {
+ LOG.info("group commit new strategy select be, tableToBeMap {},
bePressureMap {}", tableToBeMap.toString(),
+ bePressureMap.toString());
+ if (tableToBeMap.containsKey(tableId)) {
+ // todo(lyk): change 1000 to table.groupCommitSize
+ if (bePressureMap.get(tableToBeMap.get(tableId)).get() < 1000) {
+ return tableToBeMap.get(tableId);
+ } else {
+ tableToBeMap.remove(tableId);
+ }
+ }
+ Random random = new Random();
+ List<Long> backendIds =
Env.getCurrentSystemInfo().getAllBackendIds(true);
+
+ int size = backendIds.size();
+
+ while (size > 0) {
+ Long backendId = backendIds.get(random.nextInt(size));
+ Backend candidateBe =
Env.getCurrentSystemInfo().getBackend(backendId);
+
+ if (!candidateBe.isDecommissioned()) {
+ tableToBeMap.put(tableId, backendId);
+ // todo(lyk): change 10 to table.groupTime
+ bePressureMap.put(backendId, new SlidingWindowCounter(10));
+ return backendId;
+ }
+
+ backendIds.remove(backendId);
+ size = backendIds.size();
Review Comment:
random code is same as in GroupComitPlanner?
##########
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java:
##########
@@ -1603,6 +1624,20 @@ private boolean loadTxnCommitImpl(TLoadTxnCommitRequest
request) throws UserExce
request.getTbl(), request.getUserIp(),
PrivPredicate.LOAD);
}
}
+ if (request.groupCommit) {
+ String clientAddr = getClientAddrAsString();
+ SystemInfoService systemInfoService = Env.getCurrentSystemInfo();
+ long backendId = -1;
+ for (Long id : systemInfoService.getAllBackendIds(false)) {
+ Backend backend = systemInfoService.getBackend(id);
+ if (clientAddr.equals(backend.getHost())) {
+ backendId = id;
+ break;
+ }
+ }
Review Comment:
It's not a good idea to run a loop here, if there are 100 bes, then a high
frequent load may leads to a problem.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
